okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauffeat(api)!: migrate daemon and clients to protocol 361c43ca

main
branchesmainall refs →
2.2 KiB52 linesraw
1"""Forced LCM preparation preserves its newest tool unit and stored summaries."""
2
3import json
4import unittest
5
6from harness import Albedo, Provider, python, text
7
8
9class LcmPreparationTests(unittest.TestCase):
10    def test_forced_fold_retains_tool_tail_on_next_preparation(self):
11        def reply(request):
12            inputs = request["input"]
13            if "<newly-evicted-history>" in json.dumps(inputs):
14                return text("archived earlier conversation")
15            if inputs[-1].get("type") == "function_call_output":
16                return text("tool finished")
17            if inputs[-1].get("content") == "newest tool unit":
18                return python("print('TAIL_TOOL_OUTPUT')")
19            return text("ordinary answer")
20
21        provider = Provider(reply)
22        self.addCleanup(provider.close)
23        with Albedo(provider, protocol="responses") as app:
24            session = app.session()
25            for prompt in ("older unit one", "older unit two", "newest tool unit"):
26                app.prompt(session, prompt).close()
27                app.idle(session)
28            with app.api(
29                f"/sessions/{session}/compaction",
30                {"strategy": "lcm"},
31            ) as response:
32                result = json.load(response)
33            self.assertEqual(result["state"], "compacted")
34            app.idle(session)
35            app.prompt(session, "after fold").close()
36            app.idle(session)
37            sent = provider.requests[-1]["request"]["input"]
38            self.assertIn("LCM summary node #", json.dumps(sent))
39            self.assertIn("newest tool unit", json.dumps(sent))
40            self.assertIn("TAIL_TOOL_OUTPUT", json.dumps(sent))
41            calls = [
42                item["call_id"] for item in sent if item.get("type") == "function_call"
43            ]
44            outputs = [
45                item["call_id"]
46                for item in sent
47                if item.get("type") == "function_call_output"
48            ]
49            self.assertEqual(calls, outputs)
50            with app.api(f"/sessions/{session}/context") as response:
51                observation = json.load(response)["compaction"]
52            self.assertEqual(observation["strategy"], "lcm")