mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-07-09 17:08:29 +00:00
Fix Responses state after history compaction
Clear the active Responses provider continuation when automatic history compression or manual chat compaction rewrites local history, while preserving stored response IDs for cleanup. Add focused regressions for both compression paths so compacted chats do not keep stale provider-side context.
This commit is contained in:
parent
67895f7b44
commit
ad148f24cc
9 changed files with 155 additions and 4 deletions
|
|
@ -12,6 +12,7 @@
|
|||
|
||||
- Preserve history consistency before saving chats.
|
||||
- Do not skip persistence for successful loops unless the hook contract explicitly permits it.
|
||||
- History compression that rewrites local history must clear the active Responses provider continuation while preserving stored response IDs for cleanup.
|
||||
|
||||
## Work Guidance
|
||||
|
||||
|
|
|
|||
|
|
@ -1,10 +1,18 @@
|
|||
from helpers.extension import Extension
|
||||
from agent import LoopData
|
||||
from helpers.defer import DeferredTask, THREAD_BACKGROUND
|
||||
from helpers.history import clear_responses_provider_state
|
||||
|
||||
DATA_NAME_TASK = "_organize_history_task"
|
||||
|
||||
|
||||
async def compress_history(agent) -> bool:
|
||||
compressed = bool(await agent.history.compress())
|
||||
if compressed:
|
||||
clear_responses_provider_state(agent)
|
||||
return compressed
|
||||
|
||||
|
||||
class OrganizeHistory(Extension):
|
||||
async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
|
||||
if not self.agent:
|
||||
|
|
@ -17,6 +25,6 @@ class OrganizeHistory(Extension):
|
|||
|
||||
# start task
|
||||
task = DeferredTask(thread_name=THREAD_BACKGROUND)
|
||||
task.start_task(self.agent.history.compress)
|
||||
task.start_task(compress_history, self.agent)
|
||||
# set to agent to be able to wait for it
|
||||
self.agent.set_data(DATA_NAME_TASK, task)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,9 @@
|
|||
from helpers.extension import Extension
|
||||
from agent import LoopData
|
||||
from extensions.python.message_loop_end._10_organize_history import DATA_NAME_TASK
|
||||
from extensions.python.message_loop_end._10_organize_history import (
|
||||
DATA_NAME_TASK,
|
||||
compress_history,
|
||||
)
|
||||
from helpers.defer import DeferredTask, THREAD_BACKGROUND
|
||||
|
||||
MAX_SYNC_COMPRESSION_PASSES = 64
|
||||
|
|
@ -33,7 +36,7 @@ class OrganizeHistoryWait(Extension):
|
|||
else:
|
||||
# no task was running, start and wait
|
||||
self.agent.context.log.set_progress("Compressing history...")
|
||||
compressed = await self.agent.history.compress()
|
||||
compressed = await compress_history(self.agent)
|
||||
|
||||
after_tokens = self.agent.history.get_tokens()
|
||||
if not compressed or after_tokens >= before_tokens:
|
||||
|
|
|
|||
|
|
@ -723,6 +723,28 @@ def output_text(messages: list[OutputMessage], ai_label="ai", human_label="human
|
|||
return "\n".join(_stringify_output(o, ai_label, human_label) for o in messages)
|
||||
|
||||
|
||||
def clear_responses_provider_state(agent) -> None:
|
||||
key = getattr(agent, "DATA_NAME_RESPONSES_STATE", "responses_state")
|
||||
get_data = getattr(agent, "get_data", None)
|
||||
set_data = getattr(agent, "set_data", None)
|
||||
if not callable(get_data) or not callable(set_data):
|
||||
return
|
||||
|
||||
state = get_data(key)
|
||||
if not isinstance(state, dict):
|
||||
return
|
||||
|
||||
state = dict(state)
|
||||
removed = False
|
||||
for field in ("response_id", "previous_response_id"):
|
||||
if field in state:
|
||||
state.pop(field, None)
|
||||
removed = True
|
||||
|
||||
if removed:
|
||||
set_data(key, state)
|
||||
|
||||
|
||||
def _merge_outputs(a: MessageContent, b: MessageContent) -> MessageContent:
|
||||
if isinstance(a, str) and isinstance(b, str):
|
||||
return a + "\n" + b
|
||||
|
|
|
|||
|
|
@ -65,6 +65,7 @@
|
|||
- `group_messages_abab(messages: list[BaseMessage]) -> list[BaseMessage]`
|
||||
- `output_langchain(messages: list[OutputMessage])`
|
||||
- `output_text(messages: list[OutputMessage], ai_label=..., human_label=...)`
|
||||
- `clear_responses_provider_state(agent) -> None`
|
||||
- `_merge_outputs(a: MessageContent, b: MessageContent) -> MessageContent`
|
||||
- `_merge_properties(a: Dict[str, MessageContent], b: Dict[str, MessageContent]) -> Dict[str, MessageContent]`
|
||||
- `_is_raw_message(obj: object) -> bool`
|
||||
|
|
@ -77,6 +78,7 @@
|
|||
|
||||
- Helper modules own reusable framework APIs and must preserve public callers unless all callers, tests, and docs are updated together.
|
||||
- Update this file whenever public functions, classes, persistence behavior, path/security assumptions, side effects, or cross-module contracts change.
|
||||
- `clear_responses_provider_state(agent)` removes the active provider continuation IDs after local history rewrites while preserving stored response ID lists for later cleanup.
|
||||
- Observed side-effect areas: filesystem writes, filesystem deletion, model calls, plugin state, settings/state persistence, secret handling.
|
||||
- Imported dependency areas include: `abc`, `asyncio`, `collections`, `collections.abc`, `enum`, `helpers`, `json`, `langchain_core.messages`, `math`, `plugins._model_config.helpers.model_config`, `typing`, `uuid`.
|
||||
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@
|
|||
- Backup JSON and transcript artifacts must remain UTF-8 writable when chat content contains malformed Unicode such as lone surrogates.
|
||||
- Keep generated summaries bounded by configured model and token limits.
|
||||
- Preserve loaded skill names from `skill_instructions` metadata without copying full skill bodies into compacted summaries.
|
||||
- After replacing local history, clear the active Responses provider continuation while preserving stored response IDs for cleanup.
|
||||
- Do not discard original context data unless the compaction flow explicitly owns that behavior.
|
||||
|
||||
## Work Guidance
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from collections import deque
|
|||
import models as models_module
|
||||
from agent import Agent
|
||||
from helpers import files, tokens
|
||||
from helpers.history import History, output_text
|
||||
from helpers.history import History, clear_responses_provider_state, output_text
|
||||
from helpers.persist_chat import (
|
||||
export_json_chat,
|
||||
get_chat_folder_path,
|
||||
|
|
@ -145,6 +145,7 @@ async def run_compaction(
|
|||
|
||||
agent.history = History(agent=agent)
|
||||
agent.history.add_message(ai=True, content=compacted_content)
|
||||
clear_responses_provider_state(agent)
|
||||
|
||||
# Clear subordinate chain
|
||||
agent.data.pop(Agent.DATA_NAME_SUBORDINATE, None)
|
||||
|
|
|
|||
|
|
@ -43,6 +43,49 @@ class _RecordingModel:
|
|||
return f"summary-{len(self.user_messages)}", None
|
||||
|
||||
|
||||
class _CompactionHistory:
|
||||
def output(self):
|
||||
return [{"ai": False, "content": "hello"}]
|
||||
|
||||
|
||||
class _CompactionLog:
|
||||
def __init__(self):
|
||||
self.logs = []
|
||||
self.entries = []
|
||||
self.reset_called = False
|
||||
self.progress = None
|
||||
|
||||
def log(self, **kwargs):
|
||||
self.entries.append(kwargs)
|
||||
return _FakeLog()
|
||||
|
||||
def reset(self):
|
||||
self.reset_called = True
|
||||
|
||||
def set_progress(self, *args, **kwargs):
|
||||
self.progress = (args, kwargs)
|
||||
|
||||
|
||||
class _CompactionAgent:
|
||||
DATA_NAME_RESPONSES_STATE = "responses_state"
|
||||
|
||||
def __init__(self):
|
||||
self.history = _CompactionHistory()
|
||||
self.data = {
|
||||
"responses_state": {
|
||||
"response_id": "resp_current",
|
||||
"previous_response_id": "resp_previous",
|
||||
"response_ids": ["resp_previous", "resp_current"],
|
||||
}
|
||||
}
|
||||
|
||||
def get_data(self, key):
|
||||
return self.data.get(key)
|
||||
|
||||
def set_data(self, key, value):
|
||||
self.data[key] = value
|
||||
|
||||
|
||||
def test_pre_compaction_backup_sanitizes_surrogate_text(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
compactor,
|
||||
|
|
@ -114,3 +157,39 @@ async def test_large_compaction_does_not_send_unsplit_single_line_payload(monkey
|
|||
assert len(chunk_messages) > 2
|
||||
assert all(chunk_messages)
|
||||
assert all(len(message) <= 10_000 for message in chunk_messages)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_manual_compaction_clears_active_responses_state(monkeypatch):
|
||||
async def fake_single_pass(*args, **kwargs):
|
||||
return "summary"
|
||||
|
||||
agent = _CompactionAgent()
|
||||
context = SimpleNamespace(
|
||||
id="compact-chat",
|
||||
agent0=agent,
|
||||
log=_CompactionLog(),
|
||||
streaming_agent=object(),
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
compactor,
|
||||
"_build_model",
|
||||
lambda *args: ({"ctx_length": 128000}, _RecordingModel()),
|
||||
)
|
||||
monkeypatch.setattr(compactor, "_compact_single_pass", fake_single_pass)
|
||||
monkeypatch.setattr(
|
||||
compactor,
|
||||
"_save_pre_compaction_backup",
|
||||
lambda *args: {"txt": "/tmp/pre.txt"},
|
||||
)
|
||||
monkeypatch.setattr(compactor, "save_tmp_chat", lambda *args: None)
|
||||
monkeypatch.setattr(compactor, "remove_msg_files", lambda *args: None)
|
||||
monkeypatch.setattr(compactor, "mark_dirty_all", lambda *args, **kwargs: None)
|
||||
|
||||
await compactor.run_compaction(context)
|
||||
|
||||
state = agent.data["responses_state"]
|
||||
assert "response_id" not in state
|
||||
assert "previous_response_id" not in state
|
||||
assert state["response_ids"] == ["resp_previous", "resp_current"]
|
||||
|
|
|
|||
|
|
@ -45,6 +45,23 @@ class _MaxPassHistory:
|
|||
return True
|
||||
|
||||
|
||||
class _CompressOnceHistory:
|
||||
def __init__(self):
|
||||
self.compress_calls = 0
|
||||
self.tokens = 2000
|
||||
|
||||
def is_over_limit(self):
|
||||
return self.compress_calls == 0
|
||||
|
||||
def get_tokens(self):
|
||||
return self.tokens
|
||||
|
||||
async def compress(self):
|
||||
self.compress_calls += 1
|
||||
self.tokens -= 1000
|
||||
return True
|
||||
|
||||
|
||||
class _FakeLog:
|
||||
def __init__(self):
|
||||
self.entries = []
|
||||
|
|
@ -94,3 +111,20 @@ async def test_history_wait_stops_after_max_sync_compression_passes():
|
|||
f"stopped after {MAX_SYNC_COMPRESSION_PASSES} passes"
|
||||
in agent.context.log.entries[-1]["content"]
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_history_compression_clears_active_responses_state():
|
||||
agent = _FakeAgent(_CompressOnceHistory())
|
||||
agent.data["responses_state"] = {
|
||||
"response_id": "resp_current",
|
||||
"previous_response_id": "resp_previous",
|
||||
"response_ids": ["resp_previous", "resp_current"],
|
||||
}
|
||||
|
||||
await OrganizeHistoryWait(agent).execute()
|
||||
|
||||
state = agent.data["responses_state"]
|
||||
assert "response_id" not in state
|
||||
assert "previous_response_id" not in state
|
||||
assert state["response_ids"] == ["resp_previous", "resp_current"]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue