Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151"""Retrieval keeps exact counts, scoped pages, and Unicode text offsets through the daemon."""
import jsonimport unittest
from harness import Albedo, Provider, Reply, python, text
class RetrievalTests(unittest.TestCase): def setUp(self): self.code = None self.tool = None self.arguments = None
def script(request): last = request["input"][-1] if last.get("type") == "function_call_output": return text("done") if "<newly-evicted-history>" in str(last.get("content", "")): return text("fold-only Café") if self.code is not None: return python(self.code) if self.tool is not None: return Reply( "python", tool_name=self.tool, tool_arguments=self.arguments ) return text("done")
self.provider = Provider(script) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session()
def turn(self, session, prompt): before = len(self.provider.requests) self.app.prompt(session, prompt).close() self.app.idle(session) return [entry["request"] for entry in self.provider.requests[before:]]
def retrieve(self, name, arguments): self.tool, self.arguments = name, arguments requests = self.turn(self.session, "retrieve fixture") output = next( item["output"] for item in reversed(requests[-1]["input"]) if item.get("type") == "function_call_output" ) return json.loads(output)
def test_transcript_pages_count_matches_and_preserve_grapheme_offsets(self): target = self.app.session() for prompt in ["é👩💻tail Café\r", "second Café", "third Café"]: self.turn(target, prompt) self.code = f"""agent = await agents.get({target!r})found = await agent.search_messages('CAFÉ', limit=1)assert found.count == 3, foundassert len(found.rows) == 1 and found.next_offset == 1, foundseq = found.rows[0]['seq']second = await agent.search_messages('CAFÉ', offset=1, limit=1)assert second.count == 3 and second.rows[0]['seq'] > seq, secondend = await agent.search_messages('CAFÉ', offset=3, limit=1)assert end.count == 3 and end.rows == [] and end.next_offset is None, endprefix = len(f'[row #{{seq}}]\\n[user]\\n')accent = await agent.messages(seq=seq, offset=prefix, limit=1)assert accent.content == 'é', accentemoji = await agent.messages(seq=seq, offset=prefix+1, limit=1)assert emoji.content == '👩💻', emojiboundary = await agent.messages(seq=seq, offset=prefix+11, limit=1)assert boundary.content == '\\r\\n', boundaryseparator = await agent.messages(seq=seq, offset=prefix+12, limit=1)assert separator.content == '\\n', separatorbeyond = await agent.messages(seq=seq, offset=100000, limit=1)assert beyond.content == '' and beyond.next_offset is None, beyondtry: await agent.messages(seq=seq, offset=-1) assert False, 'negative offset was accepted'except AgentsError as error: assert 'nonnegative' in str(error), errorprint('RETRIEVAL_PAGES_OK')""" requests = self.turn(self.session, "verify source retrieval") output = next( item["output"] for item in reversed(requests[-1]["input"]) if item.get("type") == "function_call_output" ) self.assertIn("RETRIEVAL_PAGES_OK", output)
def test_lcm_search_pages_sources_and_nodes_independently_inside_selected_range( self, ): for prompt in ["first Café", "second Café", "third Café", "fourth Café"]: self.turn(self.session, prompt) with self.app.api( f"/sessions/{self.session}/commands", {"name": "/compact", "arguments": "lcm"}, ) as response: result = json.load(response) self.assertTrue(result["result"]["started"]) self.app.idle(self.session) listed = self.retrieve("lcm_list", {"limit": 1}) self.assertGreater(listed["total"], 0) self.assertEqual(len(listed["folds"]), 1) node = listed["folds"][0] self.assertTrue(node["frontier"]) scoped = self.retrieve( "lcm_grep", {"pattern": "CAFÉ", "summary_id": node["id"], "limit": 1} ) self.assertGreater(scoped["source_count"], 1) self.assertEqual(scoped["node_count"], 1) self.assertEqual(len(scoped["sources"]), 1) self.assertEqual(len(scoped["nodes"]), 1) self.assertEqual(scoped["next_offset"], 1) source = scoped["sources"][0] self.assertEqual(source["node_id"], node["id"]) self.assertGreaterEqual(source["seq"], node["first_seq"]) self.assertLessEqual(source["seq"], node["last_seq"]) second = self.retrieve( "lcm_grep", {"pattern": "CAFÉ", "summary_id": node["id"], "limit": 1, "offset": 1}, ) self.assertEqual(second["source_count"], scoped["source_count"]) self.assertEqual(second["node_count"], 1) self.assertEqual(len(second["sources"]), 1) self.assertEqual(second["nodes"], []) self.assertEqual(second["next_offset"], 2) summary_only = self.retrieve( "lcm_grep", {"pattern": "fold-only", "summary_id": node["id"], "limit": 1} ) self.assertEqual(summary_only["source_count"], 0) self.assertEqual(summary_only["node_count"], 1) self.assertEqual(summary_only["sources"], []) self.assertEqual(len(summary_only["nodes"]), 1) self.assertIsNone(summary_only["next_offset"]) page = self.retrieve("lcm_expand", {"id": node["id"], "limit": 1}) self.assertEqual(page["content"], "[") self.assertEqual(page["next_offset"], 1) continued = self.retrieve( "lcm_expand", {"id": node["id"], "limit": 1, "offset": 1} ) self.assertEqual(continued["content"], "s") outside = self.retrieve( "lcm_grep", {"pattern": "CAFÉ", "summary_id": node["id"], "offset": 1000} ) self.assertEqual(outside["source_count"], scoped["source_count"]) self.assertEqual(outside["sources"], []) self.assertIsNone(outside["next_offset"])