fix: archive earlier native tool results

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`.

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),

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,12 +218,19 @@ 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()
 
             runtime._execute_tool_batch(
                 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)

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:

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")

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