diff --git a/docs/tool-output-artifacts.md b/docs/tool-output-artifacts.md index 43e45c85..93ccede6 100644 --- a/docs/tool-output-artifacts.md +++ b/docs/tool-output-artifacts.md @@ -9,5 +9,8 @@ characters in the serialized JSON result, so pages can be concatenated exactly. Artifacts are private to their user, workspace, and session. Knowing an ID does not grant access from another scope. They remain available until that session or workspace is deleted, at which point the artifact rows are deleted with the -other scoped state. Small results remain inline. The tool inventory lists the -retrieval schema and bounds. +other scoped state. The most recent native tool round remains inline. Before a +later tool round is sent to the provider, older native replies become artifact references +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`. diff --git a/openkyrozen/agent/context.py b/openkyrozen/agent/context.py index 11d9c810..90e917ca 100644 --- a/openkyrozen/agent/context.py +++ b/openkyrozen/agent/context.py @@ -99,6 +99,13 @@ def _record_context_history_metrics(self, messages: list[dict[str, Any]]) -> Non retransmitted_ids = unique_history_ids & seen entry_tokens = {entry.id: max(1, (len(json.dumps(entry.payload, ensure_ascii=False)) + 3) // 4) for entry in entries if entry.id in history_ids} + if projection is not None and projection[0] == id(messages): + for index, entry_id in projection[1].items(): + if index < len(messages) and isinstance(messages[index], ToolReply): + reply = messages[index] + sent = {"call_id": reply.call_id, "name": reply.name, + "output": reply.output, "is_error": reply.is_error} + entry_tokens[entry_id] = max(1, (len(json.dumps(sent, ensure_ascii=False)) + 3) // 4) self._context_history_metrics.set({ "new_entry_count": len(new_ids), "retransmitted_entry_count": len(retransmitted_ids), diff --git a/openkyrozen/agent/engine.py b/openkyrozen/agent/engine.py index 0a1c8f25..b7f4e368 100644 --- a/openkyrozen/agent/engine.py +++ b/openkyrozen/agent/engine.py @@ -6,7 +6,7 @@ _COMPACT_CORE_TOOLS = frozenset({ "list_dir", "find_files", "search_files", "read_file", "write_file", "edit_file", "apply_patch", - "run_cmd", "discover_tools", + "run_cmd", "discover_tools", "read_tool_artifact", }) @@ -60,6 +60,31 @@ 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 + runtime = self.runtime + memory = runtime.memory_bank + task_owner = getattr(runtime.current_session, "_delegation_parent", None) or runtime + tasks = getattr(task_owner, "tasks", None) + 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, + ) + 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.", + "error": ({"code": "archived_result", "message": "Tool call failed; see artifact.", + "retriable": False, "metadata": {"artifact_id": artifact_id}} + if reply.is_error else None), + "metadata": {"artifact_id": artifact_id}, + }, reply.is_error) + def _run_native(self, turn: TurnContext) -> None: runtime = self.runtime from openkyrozen.agent.conversation import model_response_to_dict @@ -70,12 +95,18 @@ def _run_native(self, turn: TurnContext) -> None: turn.tool_calls = [] turn.tool_records = [] turn.tool_results_text = "" + try: + self._archive_replies(messages, [index for index, item in enumerate(messages) if isinstance(item, ToolReply)]) + except Exception: + turn.protocol_error_message = "Could not retain earlier tool results; no further tools were run." + return successful_operations: set[str] = set() operation_scope = str(turn.learning_run.get("run_id")) final_text = "" response = None history_tool_names: set[str] = set() observed_plan_id = None + previous_reply_indexes: list[int] = [] max_steps = getattr(runtime.execution_context, "max_steps", None) if max_steps is None: @@ -169,6 +200,7 @@ def _run_native(self, turn: TurnContext) -> None: turn.protocol_error_message = final_text break messages.append(response) + current_reply_indexes: list[int] = [] history_tool_names.update(call.name for call in response.tool_calls) calls = [{"action": call.name, "args": call.arguments, "call_id": call.id, "call_id_source": "native", @@ -186,6 +218,7 @@ def record_native_call(call, receipt): outcome = receipt.tool_result or ToolResult(data=receipt.result_data, text=receipt.result) reply_data = tool_result_to_dict(outcome) messages.append(ToolReply(call["call_id"], call["action"], reply_data, not outcome.success)) + current_reply_indexes.append(len(messages) - 1) turn.tool_results_text += f"- `{call['action']}` returned:\n{runtime._safe_fstring(prompt_result)}\n" runtime._update_tasks_panel() @@ -193,5 +226,11 @@ def record_native_call(call, receipt): calls, operation_scope=operation_scope, successful_operations=successful_operations, call_id_source="native", on_receipt=record_native_call, ) + try: + self._archive_replies(messages, previous_reply_indexes) + except Exception: + turn.protocol_error_message = "Could not retain earlier tool results; no further tools were run." + break + previous_reply_indexes = current_reply_indexes turn.final_answer = final_text or runtime._deterministic_tool_summary(turn.tool_records) diff --git a/openkyrozen/persistence/artifacts.py b/openkyrozen/persistence/artifacts.py index 50472cf2..0c0011f7 100644 --- a/openkyrozen/persistence/artifacts.py +++ b/openkyrozen/persistence/artifacts.py @@ -21,6 +21,23 @@ def create(self, content: str, *, user_id: str, workspace_id: str, session_id: s ) return artifact_id + def get_or_create(self, content: str, *, user_id: str, workspace_id: str, session_id: str) -> str: + if not all(isinstance(value, str) and value for value in (user_id, workspace_id, session_id)): + raise ValueError("Tool artifacts require a complete session scope") + with self.database._lock, self.database.connection() as db: + row = db.execute( + "SELECT id FROM tool_artifacts WHERE user_id=? AND workspace_id=? AND session_id=? AND content=? LIMIT 1", + (user_id, workspace_id, session_id, content), + ).fetchone() + if row is not None: + return row["id"] + artifact_id = f"artifact_{uuid.uuid4().hex}" + db.execute( + "INSERT INTO tool_artifacts(id,user_id,workspace_id,session_id,content,created_at) VALUES(?,?,?,?,?,?)", + (artifact_id, user_id, workspace_id, session_id, content, utc_now()), + ) + return artifact_id + def read_page(self, artifact_id: str, offset: int, limit: int, *, user_id: str, workspace_id: str, session_id: str) -> dict | None: with self.database.connection() as db: diff --git a/tests/test_agent_engine.py b/tests/test_agent_engine.py index a4910245..3aa69599 100644 --- a/tests/test_agent_engine.py +++ b/tests/test_agent_engine.py @@ -212,6 +212,17 @@ def test_complex_native_turn_can_finish_after_its_plan_completes(self): runtime.tasks.working_plan, "status", "completed") self.assertEqual(AgentEngine(runtime).run(TurnContext("Complex task")), "Done") + def test_compact_native_artifact_reader_respects_permission(self): + permitted = {"discover_tools", "read_tool_artifact"} + runtime = SimpleNamespace( + execution_context=SimpleNamespace(discovered_tools=frozenset()), + _prompt_profile=lambda: "compact", _permitted_tool_names=lambda: permitted, + ) + engine = AgentEngine(runtime) + self.assertIn("read_tool_artifact", engine._native_tool_names()) + permitted.remove("read_tool_artifact") + self.assertNotIn("read_tool_artifact", engine._native_tool_names()) + def test_compact_native_falls_back_to_permitted_inventory_without_discovery_access(self): runtime = SimpleNamespace( execution_context=SimpleNamespace(discovered_tools=frozenset()), @@ -331,6 +342,11 @@ def execute(action, arguments, **kwargs): runtime._safe_fstring = str runtime._deterministic_tool_summary = lambda records: "tool summary" runtime.tasks = SimpleNamespace(tasks=[]) + runtime.current_session = SimpleNamespace() + runtime.memory_bank = SimpleNamespace( + user_id="user", workspace_id="workspace", session_id="session", + store=SimpleNamespace(artifacts=SimpleNamespace(get_or_create=lambda *args, **kwargs: "artifact-test")), + ) runtime._complete_turn = Mock(side_effect=lambda turn: turn.final_answer) result = AgentEngine(runtime).run(TurnContext("Review Git history")) @@ -473,6 +489,32 @@ def test_native_turn_uses_typed_calls_and_correlated_results_without_legacy_loop runtime._initial_turn_response.assert_not_called() runtime._execute_action_rounds.assert_not_called() + def test_archived_failed_result_keeps_error_identity_and_scoped_full_output(self): + import json + from openkyrozen.persistence.store import EventStore + + with tempfile.TemporaryDirectory() as directory: + store = EventStore(Path(directory) / "state.sqlite3") + memory = SimpleNamespace(store=store, user_id="alice", workspace_id="project", session_id="session") + runtime = SimpleNamespace(memory_bank=memory, current_session=SimpleNamespace(), tasks=None) + original = {"data": None, "text": "failed detail", "error": {"code": "denied"}, "metadata": {}} + messages = [ToolReply("call-1", "read_file", original, True)] + AgentEngine(runtime)._archive_replies(messages, [0]) + reply = messages[0] + self.assertTrue(reply.is_error) + self.assertEqual((reply.call_id, reply.name), ("call-1", "read_file")) + artifact_id = reply.output["data"]["artifact_id"] + page = store.artifacts.read_page(artifact_id, 0, 1000, user_id="alice", + workspace_id="project", session_id="session") + self.assertEqual(json.loads(page["content"]), original) + self.assertIsNone(store.artifacts.read_page(artifact_id, 0, 1000, user_id="alice", + workspace_id="other", session_id="session")) + reopened = EventStore(Path(directory) / "state.sqlite3") + second = [ToolReply("call-1", "read_file", original, True)] + runtime.memory_bank.store = reopened + AgentEngine(runtime)._archive_replies(second, [0]) + self.assertEqual(second[0].output["data"]["artifact_id"], artifact_id) + def test_runtime_native_provider_executes_and_continues_without_text_parsing(self): from openkyrozen.app.bootstrap import build_application, build_memory from openkyrozen.tools import ToolAdapters @@ -524,7 +566,7 @@ def chat_response(self, messages, model=None, *, tools=None): continuation=provider_continuation(self, model, "fixture", {"id": "response-2"}), metadata={"provider": self.name, "model": model}, ) - if not isinstance(messages[-1], ToolReply) or messages[-1].call_id != "read-1": + if len(self.requests) == 3 and (not isinstance(messages[-1], ToolReply) or messages[-1].call_id != "read-1"): raise AssertionError("native read result was not correlated to its call") return ModelResponse( text="The file contains verified text.", finish_reason=FinishReason.FINAL, @@ -565,10 +607,36 @@ def prepare_complex(turn): patch.object(runtime, "_collect_unwrapped_tool_calls", side_effect=AssertionError("legacy parser used")): answer = runtime.chat(runtime.current_session, "Read note.txt") + older = next(item for item in provider.requests[2][0] + if isinstance(item, ToolReply) and item.call_id == "plan-1") + original = provider.requests[1][0][-1].output + self.assertNotEqual(older.output, original) + artifact_id = older.output["data"]["artifact_id"] + page = runtime.tool_registry.get_spec("read_tool_artifact").executor({"artifact_id": artifact_id}) + import json + self.assertEqual(json.loads(page.data["content"]), original) + self.assertIsNone(runtime.memory_bank.store.artifacts.read_page( + artifact_id, 0, 1000, user_id="another-user", workspace_id=runtime.tasks.workspace_id, + session_id=runtime.tasks.session_id)) + with patch.object(runtime, "_schedule_post_turn_learning"), \ + patch.object(runtime, "_prepare_turn", side_effect=prepare_complex), \ + patch.object(runtime, "_build_memory_context", return_value=packet): + next_answer = runtime.chat(runtime.current_session, "Continue") + self.assertIn("verified text", next_answer) + historical = [item for item in provider.requests[3][0] if isinstance(item, ToolReply)] + self.assertTrue(historical) + self.assertTrue(all("artifact_id" in item.output["data"] for item in historical)) + self.assertEqual(historical[0].output["data"]["artifact_id"], artifact_id) + with patch.object(runtime.memory_bank.store.artifacts, "get_or_create", side_effect=OSError("offline")), \ + patch.object(runtime, "_schedule_post_turn_learning"): + failure = runtime.chat(runtime.current_session, "Continue again") + self.assertIn("Could not retain earlier tool results", failure) + self.assertEqual(provider.calls_attempted, 4) + self.assertIn("verified text", answer) retrieve_memory.assert_called_once() - self.assertEqual(provider.calls_attempted, 3) - self.assertEqual(len(provider.requests), 3) + self.assertEqual(provider.calls_attempted, 4) + self.assertEqual(len(provider.requests), 4) self.assertTrue(all(any(packet in item.get("content", "") for item in request[0] if isinstance(item, dict)) for request in provider.requests)) self.assertEqual(provider.requests[1][0][-1].call_id, "plan-1") diff --git a/tests/test_conversation_history.py b/tests/test_conversation_history.py index b281384b..32da26e4 100644 --- a/tests/test_conversation_history.py +++ b/tests/test_conversation_history.py @@ -112,6 +112,31 @@ def fail_second_entry(value): self.assertEqual(self.history.head, original_head) self.assertEqual(self.history.display_messages(), [{"role": "user", "content": "old"}]) + def test_context_metrics_count_archived_reply_sent_to_provider(self): + from contextvars import ContextVar + from types import SimpleNamespace + from openkyrozen.agent.context import _record_context_history_metrics + from openkyrozen.providers.models import ToolReply + + entry = self.history.entry("tool.result", { + "call_id": "call-1", "name": "read_file", "output": {"text": "x" * 6000}, "is_error": False, + }) + self.history.append(entry) + messages = [ToolReply("call-1", "read_file", {"data": {"artifact_id": "artifact-1"}, + "text": "archived"})] + runtime = SimpleNamespace( + conversation_history=self.history, _conversation_turn_id=ContextVar("turn", default=None), + _context_history_projection_ids=ContextVar("projection", default=None), + _context_history_seen=ContextVar("seen", default=set()), + _context_history_metrics=ContextVar("metrics", default={}), + ) + runtime._context_history_projection_ids.set((id(messages), {0: entry.id}, (0, 1))) + _record_context_history_metrics(runtime, messages) + self.assertEqual(runtime._context_history_metrics.get()["new_entry_count"], 1) + self.assertLess(runtime._context_history_metrics.get()["new_context_tokens"], 500) + _record_context_history_metrics(runtime, messages) + self.assertLess(runtime._context_history_metrics.get()["retransmitted_history_tokens"], 500) + def test_context_metrics_drop_compacted_native_history_entries(self): from types import SimpleNamespace from openkyrozen.agent.compaction import ContextState