diff --git a/docs/tool-output-artifacts.md b/docs/tool-output-artifacts.md index 93ccede6..ebed42de 100644 --- a/docs/tool-output-artifacts.md +++ b/docs/tool-output-artifacts.md @@ -14,3 +14,9 @@ later tool round is sent to the provider, older native replies become artifact r while their call/reply identities remain intact. The tool inventory lists the retrieval schema and bounds. Access still requires permission to use `read_tool_artifact`. + +Legacy text-protocol turns also replace earlier rounds with scoped artifact +references. Their reference includes an `Action` example whose `args` is a JSON +string containing `artifact_id`, `offset`, and `limit`. Archived rounds contain +structured results, including content beyond the legacy display limit. The newest +round stays inline, and a storage failure stops the turn before another model request. diff --git a/openkyrozen/agent/action_rounds.py b/openkyrozen/agent/action_rounds.py index 800a3dd6..41e575b6 100644 --- a/openkyrozen/agent/action_rounds.py +++ b/openkyrozen/agent/action_rounds.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import uuid from typing import Any from openkyrozen.tasks.engine import canonical_status, is_complete, is_terminal @@ -8,14 +9,23 @@ def _execute_action_rounds(self, turn): from openkyrozen.agent.delegation import GUIDANCE results: list[str] = [] + full_results: list[dict[str, Any]] = [] turn.tool_records: list[dict[str, Any]] = [] successful_operations: set[str] = set() operation_scope = str(turn.learning_run.get("run_id") or uuid.uuid4().hex) + + def full_result(receipt, prompt_result): + from openkyrozen.agent.conversation import tool_result_to_dict + result = getattr(receipt, "tool_result", None) + return {"action": receipt.action, "arguments": receipt.args, + "output": tool_result_to_dict(result) if result is not None else {"text": prompt_result}} + def record_initial_call(tc, receipt): turn.tool_records.append(self._record_turn_receipt(receipt, allow_legacy_task_matching=True)) prompt_result = self._tool_result_for_prompt(receipt) self._render_tool_result(receipt.action, prompt_result) results.append(f"- `{receipt.action}({receipt.args!r})` returned:\n{self._safe_fstring(prompt_result)}") + full_results.append(full_result(receipt, prompt_result)) if self.tasks.tasks: self._update_tasks_panel() self._execute_tool_batch( @@ -41,6 +51,20 @@ def record_initial_call(tc, receipt): evolving_messages = None previous_model_reply = None previous_tool_results: list[str] = [] + previous_full_results: list[dict[str, Any]] = [] + previous_result_slot = None + + def archive_result(slot): + if slot is None: + return + message, start, output, complete = slot + content = json.dumps({"results": complete}, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + artifact_id = self._agent_engine._store_tool_output(content) + arguments = json.dumps({"artifact_id": artifact_id, "offset": 0, "limit": 1000}) + action = json.dumps({"action": "read_tool_artifact", "args": arguments}) + reference = f"Earlier tool results archived. Retrieve a page with Action: {action}" + message["content"] = message["content"][:start] + reference + message["content"][start + len(output):] + distribution_reply = None coordinator = self.subagent_manager.coordinator delegated_this_turn = coordinator and any(run["version"] != turn.agent_versions.get(run["run_id"]) @@ -205,12 +229,33 @@ def record_initial_call(tc, receipt): summary_messages.insert(2, {"role": "system", "content": memory}) if evolving_messages is not None: summary_messages = evolving_messages + try: + archive_result(previous_result_slot) + except Exception: + turn.protocol_error_message = "Could not retain earlier tool results; no further tools were run." + break summary_messages.append({"role": "assistant", "content": previous_model_reply or ""}) + previous_result_slot = None if previous_tool_results: - summary_messages.append({"role": "user", "content": "Tool results:\n" + "\n".join(previous_tool_results)}) + output = "\n".join(previous_tool_results) + message = {"role": "user", "content": "Tool results:\n" + output} + summary_messages.append(message) + previous_result_slot = (message, len("Tool results:\n"), output, previous_full_results) if error_hint or search_throttle: summary_messages.append({"role": "user", "content": error_hint + search_throttle + self._build_task_progress_hint()}) + elif turn.tool_results_text: + message = summary_messages[-1] + start = -1 + for label in ("The tools returned:\n", "Read-only inspection results:\n", "Tool results:\n"): + position = message["content"].rfind(label + turn.tool_results_text) + if position >= 0: + start = position + len(label) + break + if start < 0: + turn.protocol_error_message = "Could not retain earlier tool results; no further tools were run." + break + previous_result_slot = (message, start, turn.tool_results_text, full_results) if distribution_reply is not None: step_reply, distribution_reply = distribution_reply, None else: @@ -375,6 +420,7 @@ def record_initial_call(tc, receipt): # execute all new tool calls round_count += 1 next_results: list[str] = [] + next_full_results: list[dict[str, Any]] = [] has_errors = False def record_followup_call(tc2, receipt): nonlocal has_errors, consecutive_search_failures, total_search_calls @@ -382,6 +428,7 @@ def record_followup_call(tc2, receipt): prompt_result = self._tool_result_for_prompt(receipt) self._render_tool_result(receipt.action, prompt_result) next_results.append(f"- `{receipt.action}({receipt.args!r})` returned:\n{self._safe_fstring(prompt_result)}") + next_full_results.append(full_result(receipt, prompt_result)) if self.tasks.tasks: self._update_tasks_panel() if not receipt.success: @@ -408,8 +455,14 @@ def record_followup_call(tc2, receipt): evolving_messages = summary_messages previous_model_reply = step_reply previous_tool_results = list(next_results) + previous_full_results = list(next_full_results) # Hard limit: too many search_web calls → one last chance to complete if total_search_calls >= 7 and consecutive_search_failures >= 3: + try: + archive_result(previous_result_slot) + except Exception: + turn.protocol_error_message = "Could not retain earlier tool results; no further tools were run." + break final_msg = ( "System: Search is unavailable (7+ calls, {0} consecutive failures). " "Stop searching now. Give your **best final answer** using whatever " @@ -419,6 +472,7 @@ def record_followup_call(tc2, receipt): ).format(consecutive_search_failures) step_reply = self._call_llm_with_spinner( summary_messages + [{"role": "assistant", "content": step_reply}, + {"role": "user", "content": "Tool results:\n" + "\n".join(next_results)}, {"role": "user", "content": final_msg}] ).strip() step_reply, search_meta = self._observe_turn_response(turn, step_reply, summary_messages) diff --git a/openkyrozen/agent/engine.py b/openkyrozen/agent/engine.py index b7f4e368..b41d6a94 100644 --- a/openkyrozen/agent/engine.py +++ b/openkyrozen/agent/engine.py @@ -60,22 +60,23 @@ def _native_tool_names(self) -> set[str]: discovered = getattr(runtime.execution_context, "discovered_tools", frozenset()) return permitted & (_COMPACT_CORE_TOOLS | set(discovered)) - def _archive_replies(self, messages, indexes): - if not indexes: - return - import json + def _store_tool_output(self, content): runtime = self.runtime memory = runtime.memory_bank task_owner = getattr(runtime.current_session, "_delegation_parent", None) or runtime tasks = getattr(task_owner, "tasks", None) + return memory.store.artifacts.get_or_create( + content, user_id=getattr(tasks, "user_id", None) or memory.user_id, + workspace_id=getattr(tasks, "workspace_id", None) or memory.workspace_id, + session_id=getattr(tasks, "session_id", None) or memory.session_id, + ) + + def _archive_replies(self, messages, indexes): + import json for index in indexes: reply = messages[index] content = json.dumps(reply.output, ensure_ascii=False, sort_keys=True, separators=(",", ":")) - artifact_id = memory.store.artifacts.get_or_create( - content, user_id=getattr(tasks, "user_id", None) or memory.user_id, - workspace_id=getattr(tasks, "workspace_id", None) or memory.workspace_id, - session_id=getattr(tasks, "session_id", None) or memory.session_id, - ) + artifact_id = self._store_tool_output(content) summary = {"artifact_id": artifact_id, "total_chars": len(content)} messages[index] = ToolReply(reply.call_id, reply.name, { "data": summary, "text": "Earlier tool result archived; use read_tool_artifact to retrieve it.", diff --git a/openkyrozen/agent/runtime.py b/openkyrozen/agent/runtime.py index d1813923..55d15e42 100644 --- a/openkyrozen/agent/runtime.py +++ b/openkyrozen/agent/runtime.py @@ -610,6 +610,11 @@ def _stop_process(self, args): def _read_tool_artifact(self, args): """Read a session-owned tool artifact in pages up to 1,000 characters.""" from openkyrozen.tools.models import ToolResult, tool_failure + if isinstance(args, str): + try: + args = json.loads(args) + except (TypeError, ValueError): + args = None artifact_id = args.get("artifact_id") if isinstance(args, dict) else None offset = args.get("offset", 0) if isinstance(args, dict) else 0 limit = args.get("limit", 1000) if isinstance(args, dict) else 1000 diff --git a/tests/test_legacy_tool_artifacts.py b/tests/test_legacy_tool_artifacts.py new file mode 100644 index 00000000..9fbe9399 --- /dev/null +++ b/tests/test_legacy_tool_artifacts.py @@ -0,0 +1,211 @@ +import copy +import json +import os +import re +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from openkyrozen.agent.turn import TurnContext +from openkyrozen.app.bootstrap import build_application, build_memory +from openkyrozen.tools import ToolAdapters + + +class LegacyToolArtifactsTests(unittest.TestCase): + def test_archived_legacy_result_retains_content_beyond_display_limit(self): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_APPROVAL_MODE": "never", + "KYROZEN_WORKSPACE_ROOT": directory, + }): + root = Path(directory) + for index in range(250): + (root / f"file_{index:04d}_long_name.txt").touch() + (root / "note.txt").write_text("note", encoding="utf-8") + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + self.addCleanup(app.close) + runtime = app.runtime + turn = TurnContext("Inspect files", interaction_mode="agent", complexity="simple", + learning_run={"run_id": "large-test"}, memory_packet="", + response_text='Action: {"action":"list_dir","args":"."}', + tool_calls=[{"action": "list_dir", "args": "."}]) + turn.agent_versions = {} + requests = [] + def response(messages): + requests.append(copy.deepcopy(messages)) + return ('Action: {"action":"read_file","args":"note.txt"}' if len(requests) == 1 + else "Files inspected.") + with patch.object(runtime, "_call_llm_with_spinner", side_effect=response), \ + patch.object(runtime, "_update_tasks_panel"): + runtime._execute_action_rounds(turn) + second = "\n".join(item["content"] for item in requests[1]) + artifact_id = re.search(r"artifact_[0-9a-f]+", second).group() + pages, offset = [], 0 + while True: + result = runtime.tool_registry.get_spec("read_tool_artifact").executor( + {"artifact_id": artifact_id, "offset": offset, "limit": 1000}).data + pages.append(result["content"]) + if result["next_offset"] is None: + break + offset = result["next_offset"] + self.assertIn("file_0249_long_name.txt", "".join(pages)) + + def test_search_limit_synthesis_uses_latest_result_and_archives_earlier_one(self): + from types import SimpleNamespace + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_APPROVAL_MODE": "never", + "KYROZEN_WORKSPACE_ROOT": directory, + }): + root = Path(directory) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + self.addCleanup(app.close) + runtime = app.runtime + turn = TurnContext("Search", interaction_mode="agent", complexity="simple", + learning_run={"run_id": "search-test"}, memory_packet="", + response_text='Action: {"action":"search_web","args":"0"}', + tool_calls=[{"action": "search_web", "args": "0"}]) + turn.agent_versions = {} + requests = [] + def response(messages): + requests.append(copy.deepcopy(messages)) + number = len(requests) + return (f'Action: {{"action":"search_web","args":"{number}"}}' + if number <= 6 else "Search failed; no results verified.") + def batch(calls, **kwargs): + for call in calls: + kwargs["on_receipt"](call, SimpleNamespace(action="search_web", args=call["args"], success=False)) + with patch.object(runtime, "_call_llm_with_spinner", side_effect=response), \ + patch.object(runtime, "_execute_tool_batch", side_effect=batch), \ + patch.object(runtime, "_record_turn_receipt", return_value={"action": "search_web", "success": False}), \ + patch.object(runtime, "_tool_result_for_prompt", side_effect=lambda receipt: "failure-" + receipt.args), \ + patch.object(runtime, "_render_tool_result"), patch.object(runtime, "_render_warning"), \ + patch.object(runtime, "_update_tasks_panel"): + runtime._execute_action_rounds(turn) + final = "\n".join(item["content"] for item in requests[-1]) + self.assertNotIn("failure-5", final) + self.assertIn("failure-6", final) + + def test_legacy_archival_failure_stops_before_next_model_request(self): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_APPROVAL_MODE": "never", + "KYROZEN_WORKSPACE_ROOT": directory, + }): + root = Path(directory) + (root / "first.txt").write_text("first", encoding="utf-8") + (root / "second.txt").write_text("second", encoding="utf-8") + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + self.addCleanup(app.close) + runtime = app.runtime + turn = TurnContext("Read both", interaction_mode="agent", complexity="simple", + learning_run={"run_id": "legacy-test"}, memory_packet="", + response_text='Action: {"action":"read_file","args":"first.txt"}', + tool_calls=[{"action": "read_file", "args": "first.txt"}]) + turn.agent_versions = {} + with patch.object(runtime, "_call_llm_with_spinner", return_value= + 'Action: {"action":"read_file","args":"second.txt"}') as call, \ + patch.object(runtime._agent_engine, "_store_tool_output", side_effect=OSError("disk full")), \ + patch.object(runtime, "_update_tasks_panel"): + runtime._execute_action_rounds(turn) + self.assertEqual(call.call_count, 1) + self.assertIn("Could not retain earlier tool results", turn.protocol_error_message) + self.assertEqual(len(turn.tool_records), 2) + + def test_legacy_artifact_read_accepts_json_and_enforces_scope(self): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_APPROVAL_MODE": "never", + "KYROZEN_WORKSPACE_ROOT": directory, + }): + root = Path(directory) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + self.addCleanup(app.close) + runtime = app.runtime + scope = {"user_id": runtime.tasks.user_id, "workspace_id": runtime.tasks.workspace_id, + "session_id": runtime.tasks.session_id} + artifact_id = runtime.memory_bank.store.artifacts.create("abcdef", **scope) + result = runtime._run_tool("read_tool_artifact", '{"artifact_id":"' + artifact_id + '","offset":2,"limit":3}', + return_receipt=True) + self.assertTrue(result.success) + self.assertEqual(result.tool_result.data["content"], "cde") + hidden = runtime.memory_bank.store.artifacts.create("private", **{**scope, "user_id": "bob"}) + denied = runtime._run_tool("read_tool_artifact", '{"artifact_id":"' + hidden + '"}', return_receipt=True) + self.assertFalse(denied.success) + self.assertNotIn("private", denied.result) + + def test_legacy_rounds_archive_results_in_each_prompt_mode(self): + for mode, profile in (("agent", "full"), ("agent", "compact"), ("ask", "full"), ("plan", "full")): + with self.subTest(mode=mode, profile=profile): + self._verify_legacy_rounds(mode, profile) + + def _verify_legacy_rounds(self, mode, profile): + with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_APPROVAL_MODE": "never", + "KYROZEN_WORKSPACE_ROOT": directory, + }): + root = Path(directory) + (root / "first.txt").write_text("FIRST_UNIQUE_CONTENT", encoding="utf-8") + (root / "second.txt").write_text("SECOND_UNIQUE_CONTENT", encoding="utf-8") + (root / "third.txt").write_text("THIRD_UNIQUE_CONTENT", encoding="utf-8") + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + self.addCleanup(app.close) + runtime = app.runtime + turn = TurnContext("Read both files", interaction_mode=mode, complexity="simple", + learning_run={"run_id": "legacy-test"}, memory_packet="", + response_text='Action: {"action":"read_file","args":"first.txt"}', + tool_calls=[{"action": "read_file", "args": "first.txt"}]) + turn.agent_versions = {} + requests = [] + final = ('PlanProposal:\n```json\n{"title":"Read files","summary":"Use the inspected files",' + '"assumptions":[],"steps":[{"id":"step-1","title":"Review",' + '"description":"Review the files","acceptance":["Reviewed"]}]}\n```' + if mode == "plan" else "All files were read.") + replies = iter(['Action: {"action":"read_file","args":"second.txt"}', + 'Action: {"action":"read_file","args":"third.txt"}']) + def response(messages): + requests.append(copy.deepcopy(messages)) + if len(requests) <= 2: + return next(replies) + if len(requests) == 3: + text = "\n".join(item["content"] for item in messages) + artifact_id = re.search(r"artifact_[0-9a-f]+", text).group() + args = json.dumps({"artifact_id": artifact_id, "offset": 0, "limit": 1000}) + return "Action: " + json.dumps({"action": "read_tool_artifact", "args": args}) + return final + with patch.object(runtime, "_call_llm_with_spinner", side_effect=response), \ + patch.object(runtime, "_prompt_profile", return_value=profile), \ + patch.object(runtime, "_update_tasks_panel"): + runtime._execute_action_rounds(turn) + self.assertEqual(len(requests), 4) + first = "\n".join(item["content"] for item in requests[0]) + second = "\n".join(item["content"] for item in requests[1]) + self.assertIn("FIRST_UNIQUE_CONTENT", first) + self.assertNotIn("FIRST_UNIQUE_CONTENT", second) + self.assertIn("SECOND_UNIQUE_CONTENT", second) + third = "\n".join(item["content"] for item in requests[2]) + self.assertNotIn("FIRST_UNIQUE_CONTENT", third) + self.assertNotIn("SECOND_UNIQUE_CONTENT", third) + self.assertIn("THIRD_UNIQUE_CONTENT", third) + fourth = "\n".join(item["content"] for item in requests[3]) + self.assertIn("FIRST_UNIQUE_CONTENT", fourth) + self.assertNotIn("THIRD_UNIQUE_CONTENT", fourth) + artifact_id = re.search(r"artifact_[0-9a-f]+", second).group() + receipt = runtime._run_tool("read_tool_artifact", '{"artifact_id":"' + artifact_id + '"}', + return_receipt=True) + self.assertTrue(receipt.success) + self.assertIn("FIRST_UNIQUE_CONTENT", receipt.result) + reopened = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + self.addCleanup(reopened.close) + reopened.runtime.bind_interaction_scope(runtime.tasks.session_id, user_id="alice") + restored = reopened.runtime._run_tool("read_tool_artifact", '{"artifact_id":"' + artifact_id + '"}', + return_receipt=True) + self.assertTrue(restored.success) + self.assertIn("FIRST_UNIQUE_CONTENT", restored.result) + + +if __name__ == "__main__": + unittest.main()