feat(session): enforce replay/file-cap invariants for history lifecycle

2026-04-27 00:53:32 +08:00
parent c64ec3e73c
commit 59dfd74842
8 changed files with 374 additions and 14 deletions
@@ -269,3 +269,66 @@ def test_get_history_ignores_media_kwarg_on_non_user_rows():
    # List content is passed through verbatim — the synthesizer only
    # rewrites plain-string content.
    assert history[0]["content"] == [{"type": "text", "text": "structured"}]
+
+
+def test_get_history_respects_max_tokens(monkeypatch):
+    session = Session(key="test:token-cap")
+    session.messages.extend(
+        [
+            {"role": "user", "content": "u1"},
+            {"role": "assistant", "content": "a1"},
+            {"role": "user", "content": "u2"},
+            {"role": "assistant", "content": "a2"},
+            {"role": "user", "content": "u3"},
+            {"role": "assistant", "content": "a3"},
+        ]
+    )
+
+    token_map = {"u1": 50, "a1": 50, "u2": 50, "a2": 50, "u3": 50, "a3": 50}
+    monkeypatch.setattr(
+        "nanobot.session.manager.estimate_message_tokens",
+        lambda message: token_map.get(message.get("content"), 0),
+    )
+
+    history = session.get_history(max_messages=500, max_tokens=120)
+    assert [m["content"] for m in history] == ["u3", "a3"]
+
+
+def test_get_history_recovers_user_when_token_slice_would_be_assistant_only(monkeypatch):
+    session = Session(key="test:assistant-only-slice")
+    session.messages.extend(
+        [
+            {"role": "user", "content": "u1"},
+            {"role": "assistant", "content": "a1"},
+            {"role": "user", "content": "u2"},
+            {"role": "assistant", "content": "a2"},
+        ]
+    )
+    token_map = {"u1": 100, "a1": 100, "u2": 100, "a2": 100}
+    monkeypatch.setattr(
+        "nanobot.session.manager.estimate_message_tokens",
+        lambda message: token_map.get(message.get("content"), 0),
+    )
+
+    history = session.get_history(max_messages=500, max_tokens=100)
+    assert [m["content"] for m in history] == ["u2", "a2"]
+
+
+def test_retain_recent_legal_suffix_hard_cap_with_long_non_user_chain():
+    session = Session(key="test:hard-cap-chain")
+    session.messages.append({"role": "user", "content": "u0"})
+    session.messages.append(
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [
+                {"id": "c1", "type": "function", "function": {"name": "x", "arguments": "{}"}}
+            ],
+        }
+    )
+    for i in range(12):
+        session.messages.append({"role": "assistant", "content": f"a{i}"})
+
+    session.retain_recent_legal_suffix(6)
+
+    assert len(session.messages) <= 6