repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
Mirrored from https://github.com/yingqi-z20/Agent-libOS
stars
latest
clone command
git clone gitlawb://did:key:z6MkqRzA...RfoM/yingqi-z20-Agen...git clone gitlawb://did:key:z6MkqRzA.../yingqi-z20-Agen...d98dd2c9IPC1d ago| #1 | from __future__ import annotations |
| #2 | |
| #3 | import json |
| #4 | import tempfile |
| #5 | import unittest |
| #6 | from types import SimpleNamespace |
| #7 | from typing import Any |
| #8 | |
| #9 | from agent_libos import Runtime |
| #10 | from agent_libos.llm.client import LLMCompletion |
| #11 | from agent_libos.llm.context_memory import context_object_name |
| #12 | from agent_libos.models import ObjectRight |
| #13 | |
| #14 | |
| #15 | class LLMContextMemoryTests(unittest.TestCase): |
| #16 | def test_llm_context_is_process_readable_writable_memory_object(self) -> None: |
| #17 | runtime = Runtime.open("local") |
| #18 | try: |
| #19 | runtime.llm.client = RecordingActionClient( |
| #20 | [{"action": "create_memory_object", "type": "observation", "payload": {"seen": 1}}] |
| #21 | ) |
| #22 | pid = runtime.process.spawn(image="base-agent:v0", goal="create context") |
| #23 | |
| #24 | runtime.run_next_process_once() |
| #25 | |
| #26 | name = context_object_name(pid) |
| #27 | obj = runtime.store.get_object_by_name(name, namespace=runtime.memory.resolve_namespace(pid)) |
| #28 | self.assertIsNotNone(obj) |
| #29 | assert obj is not None |
| #30 | self.assertFalse(obj.immutable) |
| #31 | self.assertEqual(obj.payload["kind"], "llm_context") |
| #32 | self.assertTrue(runtime.capability.check(pid, f"object:{obj.oid}", ObjectRight.READ)) |
| #33 | self.assertTrue(runtime.capability.check(pid, f"object:{obj.oid}", ObjectRight.WRITE)) |
| #34 | process = runtime.process.get(pid) |
| #35 | self.assertIn(obj.oid, [handle.oid for handle in process.memory_view.roots]) |
| #36 | |
| #37 | read = runtime.tools.call(pid, "read_memory_object", {"name": name}) |
| #38 | appended = runtime.tools.call( |
| #39 | pid, |
| #40 | "append_memory_object", |
| #41 | {"name": name, "entry": {"kind": "agent_note", "text": "keep this in context"}}, |
| #42 | ) |
| #43 | |
| #44 | updated = runtime.store.get_object_by_name(name, namespace=runtime.memory.resolve_namespace(pid)) |
| #45 | self.assertTrue(read.ok, read.error) |
| #46 | self.assertTrue(appended.ok, appended.error) |
| #47 | self.assertEqual(updated.payload["entries"][-1]["kind"], "agent_note") |
| #48 | finally: |
| #49 | runtime.close() |
| #50 | |
| #51 | def test_llm_context_prompt_grows_by_appending_to_preserve_cache_prefix(self) -> None: |
| #52 | runtime = Runtime.open("local") |
| #53 | try: |
| #54 | runtime.llm.client = RecordingActionClient( |
| #55 | [ |
| #56 | {"action": "create_memory_object", "type": "observation", "payload": {"step": 1}}, |
| #57 | {"action": "create_memory_object", "type": "observation", "payload": {"step": 2}}, |
| #58 | ] |
| #59 | ) |
| #60 | pid = runtime.process.spawn(image="base-agent:v0", goal="append context") |
| #61 | |
| #62 | runtime.run_next_process_once() |
| #63 | runtime.run_next_process_once() |
| #64 | |
| #65 | first, second = runtime.llm.client.user_prompts |
| #66 | self.assertIn("Cache strategy: append_only_stable_prefix", first) |
| #67 | self.assertIn("LLM context object", first) |
| #68 | self.assertTrue(second.startswith(first)) |
| #69 | context = runtime.store.get_object_by_name( |
| #70 | context_object_name(pid), |
| #71 | namespace=runtime.memory.resolve_namespace(pid), |
| #72 | ) |
| #73 | kinds = [entry["kind"] for entry in context.payload["entries"]] |
| #74 | self.assertIn("memory_delta", kinds) |
| #75 | self.assertGreater(len(second), len(first)) |
| #76 | finally: |
| #77 | runtime.close() |
| #78 | |
| #79 | def test_llm_prompt_lists_only_process_visible_tools(self) -> None: |
| #80 | runtime = Runtime.open("local") |
| #81 | try: |
| #82 | runtime.llm.client = RecordingActionClient( |
| #83 | [{"action": "process_exit", "payload": {"done": True}}] |
| #84 | ) |
| #85 | pid = runtime.process.spawn(image="base-agent:v0", goal="exit") |
| #86 | |
| #87 | runtime.run_next_process_once() |
| #88 | |
| #89 | tool_names = { |
| #90 | tool["function"]["name"] |
| #91 | for tool in runtime.llm.client.tool_batches[0] |
| #92 | } |
| #93 | self.assertIn("process_exit", tool_names) |
| #94 | self.assertNotIn("read_text_file", tool_names) |
| #95 | self.assertNotIn("read_text_file", runtime.llm.client.user_prompts[0]) |
| #96 | finally: |
| #97 | runtime.close() |
| #98 | |
| #99 | def test_llm_retries_malformed_empty_tool_name_once(self) -> None: |
| #100 | runtime = Runtime.open("local") |
| #101 | try: |
| #102 | runtime.llm.client = RecordingActionClient( |
| #103 | [ |
| #104 | {"action": "", "path": "."}, |
| #105 | {"action": "process_exit", "payload": {"done": True}}, |
| #106 | ] |
| #107 | ) |
| #108 | pid = runtime.process.spawn(image="base-agent:v0", goal="recover malformed action") |
| #109 | |
| #110 | result = runtime.run_next_process_once() |
| #111 | |
| #112 | self.assertTrue(result["ok"]) |
| #113 | self.assertEqual(result["action"]["action"], "process_exit") |
| #114 | self.assertEqual(len(runtime.llm.client.user_prompts), 2) |
| #115 | self.assertIn("could not be dispatched", runtime.llm.client.user_prompts[1]) |
| #116 | repairs = [record for record in runtime.audit.trace() if record.action == "llm.action_repair_requested"] |
| #117 | self.assertEqual(len(repairs), 1) |
| #118 | assert repairs[0].decision is not None |
| #119 | self.assertEqual(repairs[0].decision["tool_calls_preview"][0]["name"], "") |
| #120 | self.assertIn('"path"', repairs[0].decision["tool_calls_preview"][0]["arguments_preview"]) |
| #121 | finally: |
| #122 | runtime.close() |
| #123 | |
| #124 | def test_llm_call_records_persist_prompt_output_usage_and_reasoning(self) -> None: |
| #125 | with tempfile.TemporaryDirectory() as temp_dir: |
| #126 | db = f"{temp_dir}/runtime.sqlite" |
| #127 | runtime = Runtime.open(db) |
| #128 | try: |
| #129 | runtime.llm.client = MetadataActionClient() |
| #130 | pid = runtime.process.spawn(image="base-agent:v0", goal="persist llm calls") |
| #131 | |
| #132 | runtime.run_next_process_once() |
| #133 | |
| #134 | calls = runtime.store.list_llm_calls(pid) |
| #135 | |
| #136 | self.assertEqual(len(calls), 1) |
| #137 | call = calls[0] |
| #138 | self.assertEqual(call.pid, pid) |
| #139 | self.assertEqual(call.purpose, "action_selection") |
| #140 | self.assertEqual(call.status, "ok") |
| #141 | self.assertEqual(call.api, "chat") |
| #142 | self.assertEqual(call.model, "test-model") |
| #143 | self.assertEqual(call.request_id, "req_123") |
| #144 | self.assertEqual(call.response_id, "resp_123") |
| #145 | self.assertEqual(call.response_content, "visible assistant text") |
| #146 | self.assertEqual(call.usage["total_tokens"], 17) |
| #147 | self.assertEqual(call.reasoning, {"summary": "selected process_exit"}) |
| #148 | self.assertEqual(call.raw_response["id"], "raw_resp") |
| #149 | self.assertEqual(call.tool_calls[0]["name"], "process_exit") |
| #150 | self.assertIn("persist llm calls", call.messages[1]["content"]) |
| #151 | self.assertTrue(any(tool["function"]["name"] == "process_exit" for tool in call.tools)) |
| #152 | finally: |
| #153 | runtime.close() |
| #154 | |
| #155 | reopened = Runtime.open(db) |
| #156 | try: |
| #157 | persisted = reopened.store.list_llm_calls() |
| #158 | |
| #159 | self.assertEqual(len(persisted), 1) |
| #160 | self.assertEqual(persisted[0].usage["prompt_tokens"], 13) |
| #161 | self.assertEqual(persisted[0].reasoning, {"summary": "selected process_exit"}) |
| #162 | finally: |
| #163 | reopened.close() |
| #164 | |
| #165 | |
| #166 | class RecordingActionClient: |
| #167 | def __init__(self, actions: list[dict[str, Any]]): |
| #168 | self.actions = list(actions) |
| #169 | self.user_prompts: list[str] = [] |
| #170 | self.tool_batches: list[list[dict[str, Any]]] = [] |
| #171 | |
| #172 | def complete_action(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> LLMCompletion: |
| #173 | self.user_prompts.append(str(messages[-1]["content"])) |
| #174 | self.tool_batches.append(tools) |
| #175 | action = self.actions.pop(0) |
| #176 | name = str(action["action"]) |
| #177 | args = {key: value for key, value in action.items() if key != "action"} |
| #178 | return LLMCompletion( |
| #179 | content="", |
| #180 | tool_calls=[{"id": f"context_{len(self.user_prompts)}", "name": name, "arguments": json.dumps(args)}], |
| #181 | ) |
| #182 | |
| #183 | |
| #184 | class MetadataActionClient: |
| #185 | def complete_action(self, messages: list[dict[str, Any]], tools: list[dict[str, Any]]) -> LLMCompletion: |
| #186 | return LLMCompletion( |
| #187 | content="visible assistant text", |
| #188 | tool_calls=[ |
| #189 | { |
| #190 | "id": "tool_123", |
| #191 | "name": "process_exit", |
| #192 | "arguments": json.dumps({"payload": {"done": True}}), |
| #193 | } |
| #194 | ], |
| #195 | raw=SimpleNamespace(id="raw_resp", provider="fake"), |
| #196 | api="chat", |
| #197 | response_id="resp_123", |
| #198 | request_id="req_123", |
| #199 | model="test-model", |
| #200 | usage={"prompt_tokens": 13, "completion_tokens": 4, "total_tokens": 17}, |
| #201 | reasoning={"summary": "selected process_exit"}, |
| #202 | ) |
| #203 | |
| #204 | |
| #205 | if __name__ == "__main__": |
| #206 | unittest.main() |
| #207 |