From 13f693b3438ecf3b9eddccb06ea20d55cf10f614 Mon Sep 17 00:00:00 2001 From: Kai Date: Sun, 25 Oct 2026 22:12:12 +0800 Subject: [PATCH] fix(storage): preserve developer messages in session snapshots Validate developer history as instruction content during snapshot and journal loading. Preserve its role through autosave, exports, and resumed Responses requests. Cover native SDK continuation, saved identity and usage, legacy formats, and malformed message rejection. --- opencollab/adapters/storage.py | 4 +- .../test_developer_session_persistence.py | 159 ++++++++++++++++++ 2 files changed, 161 insertions(+), 2 deletions(-) create mode 100644 tests/runtime/test_developer_session_persistence.py diff --git a/opencollab/adapters/storage.py b/opencollab/adapters/storage.py index b9da8a7d..e2f60f41 100644 --- a/opencollab/adapters/storage.py +++ b/opencollab/adapters/storage.py @@ -141,7 +141,7 @@ def write(self, text: str) -> int: return len(text) class SessionStore: - allowed_roles = {"system", "user", "assistant", "tool"} + allowed_roles = {"system", "developer", "user", "assistant", "tool"} def has_snapshot(self, path: str) -> bool: """Return whether a base or recoverable journal sidecar is present.""" @@ -296,7 +296,7 @@ def _validate_messages(self, messages: list[dict[str, Any]]) -> None: role = msg.get("role") if role not in self.allowed_roles: raise ValueError(f"Invalid message role at position {lineno}: {role}") - if role in {"system", "user"}: + if role in {"system", "developer", "user"}: self._validate_content(msg, lineno) elif role == "assistant": tool_calls = msg.get("tool_calls") diff --git a/tests/runtime/test_developer_session_persistence.py b/tests/runtime/test_developer_session_persistence.py new file mode 100644 index 00000000..d7abb53d --- /dev/null +++ b/tests/runtime/test_developer_session_persistence.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import json + +import pytest + +from opencollab.adapters.llm.client import LLMClient +from opencollab.adapters.storage import SessionStore +from opencollab.bootstrap import build_session, load_session +from opencollab.domain.agent import Agent +from opencollab.domain.session import SessionPhase +from opencollab.sdk import OpenCollab +from tests.support.environment_test_support import local_environment_factory as local_environment_factory +from tests.support.provider_sdk_http import http_response, install_sdk_transport +from tests.support.responses_provider_test_support import message_item + + +@pytest.mark.parametrize("content", ["Use a concise answer.", [{"type": "text", "text": "Use a concise answer."}]]) +@pytest.mark.parametrize("saved", ["autosave", "checkpoint", "export"]) +async def test_responses_developer_history_survives_save_and_next_turn( + local_environment_factory, monkeypatch, tmp_path, content, saved, +): + usage_path = tmp_path / "usage.jsonl" + monkeypatch.setenv("OPENCOLLAB_API_USAGE_LOG", str(usage_path)) + monkeypatch.setenv("OPENCOLLAB_BUDGET_NUDGE_MODE", "off") + requests = [] + + def handler(request): + body = json.loads(request.content) + requests.append(body) + return http_response(request, 200, json={ + "id": f"resp_{len(requests)}", "object": "response", "created_at": 1, + "status": "completed", "error": None, "incomplete_details": None, + "model": body["model"], "output": [message_item(f"answer {len(requests)}")], + "usage": {"input_tokens": 10, "output_tokens": 2, "total_tokens": 12}, + }) + + install_sdk_transport(monkeypatch, "openai", handler) + automatic = tmp_path / "automatic.json" + path = tmp_path / "exported.json" if saved == "export" else automatic + agent = Agent(name="lead", system_prompt="System instructions.", model="o1-pro", wire_protocol="responses") + async with LLMClient( + model="o1-pro", wire_protocol="responses", max_retries=0, + api_key="controlled-test", # pragma: allowlist secret + base_url="https://controlled.invalid/v1", + ) as client: + original = build_session( + agent=agent, env=local_environment_factory(), llm=client, + auto_save_path=str(automatic), aid=7, run_id="developer-roundtrip", + ) + restored = None + try: + original.messages = [*original.messages, {"role": "developer", "content": content}] + await original.add_user_message("First request.") + assert await original.run_loop() == "answer 1" + assert original.persistence_errors == () + assert original.phase is SessionPhase.DONE + assert original.used_tokens == 12 + if saved != "autosave": + original.save(str(path)) + snapshot = OpenCollab.read_session_snapshot(path) + assert snapshot["messages"][1]["role"] == "developer" + assert snapshot["messages"][1]["content"] == content + assert snapshot["aid"] == 7 + assert snapshot["run_id"] == "developer-roundtrip" + assert snapshot["session_state"]["used_tokens"] == 12 + assert snapshot["session_state"]["phase"] == "done" + messages = original.messages.copy() + await original.aclose() + + restored = load_session( + str(path), agent=agent, env=local_environment_factory(), llm=client, + auto_save_path=str(automatic), + ) + assert restored.messages == messages + assert restored.state.aid == 7 + assert restored.run_id == "developer-roundtrip" + assert restored.used_tokens == 12 + assert restored.phase is SessionPhase.DONE + await restored.add_user_message("Second request.") + assert await restored.run_loop() == "answer 2" + assert restored.used_tokens == 24 + assert restored.persistence_errors == () + persisted = OpenCollab.read_session_snapshot(automatic) + assert persisted["session_state"]["used_tokens"] == 24 + assert persisted["aid"] == 7 + assert persisted["run_id"] == "developer-roundtrip" + assert persisted["messages"][1]["role"] == "developer" + assert persisted["messages"][1]["content"] == content + finally: + await original.aclose() + if restored is not None: + await restored.aclose() + + assert len(requests) == 2 + expected = content if isinstance(content, str) else [{"type": "input_text", "text": content[0]["text"]}] + for request in requests: + assert request["instructions"] == "System instructions." + assert request["input"][0] == {"role": "developer", "content": expected} + assert requests[0]["input"][1] == {"role": "user", "content": "First request."} + assert requests[1]["input"][-1] == {"role": "user", "content": "Second request."} + assert [row["role"] for row in requests[1]["input"]] == ["developer", "user", "assistant", "user"] + usage = [json.loads(line) for line in usage_path.read_text(encoding="utf-8").splitlines()] + assert len(usage) == 2 + assert [row["usage"]["total_tokens"] for row in usage] == [12, 12] + + +@pytest.mark.parametrize("format", ["json", "legacy-json", "jsonl", "journal", "checkpoint"]) +@pytest.mark.parametrize("role", ["developer", "system", "user"]) +def test_instruction_roles_round_trip_through_store_formats(tmp_path, format, role): + path = tmp_path / "session.json" + messages = [ + {"role": "system", "content": "System instructions."}, + {"role": role, "content": [{"type": "input_text", "text": "Keep this instruction."}]}, + ] + store = SessionStore() + if format == "json": + store.save(str(path), messages, meta={"snapshot_version": 1, "aid": 7}) + elif format == "legacy-json": + path.write_text(json.dumps(messages), encoding="utf-8") + elif format == "jsonl": + path.write_text("".join(json.dumps(message) + "\n" for message in messages), encoding="utf-8") + elif format == "journal": + store.append_snapshot_delta(str(path), sequence=1, replace_from=0, messages=messages, meta={}) + else: + store.append_snapshot_delta(str(path), sequence=1, replace_from=0, messages=messages, meta={}) + store.checkpoint_snapshot(str(path), messages, sequence=1, meta={}) + assert store.load_messages(str(path), "fallback") == messages + + +@pytest.mark.parametrize("format", ["json", "jsonl", "journal"]) +@pytest.mark.parametrize(("message", "error"), [ + ({"role": "developer"}, "missing 'content'"), + ({"role": "developer", "content": None}, "must be text or a content-part list"), + ({"role": "developer", "content": {}}, "must be text or a content-part list"), + ({"role": "developer", "content": []}, "part list must not be empty"), + ({"role": "developer", "content": [{"type": "input_text"}]}, "requires text"), + ({"role": "developer", "content": [{"type": "text", "text": 42}]}, "requires text"), + ({"role": "unknown", "content": "invalid"}, "Invalid message role"), + ({"role": "Developer", "content": "invalid"}, "Invalid message role"), + ({"role": "tool", "content": "result"}, "expected non-empty 'tool_call_id'"), +]) +def test_store_still_rejects_malformed_messages(tmp_path, format, message, error): + path = tmp_path / "invalid.json" + store = SessionStore() + if format == "json": + path.write_text(json.dumps({"messages": [message]}), encoding="utf-8") + elif format == "jsonl": + path.write_text( + json.dumps({"role": "system", "content": "System instructions."}) + "\n" + json.dumps(message) + "\n", + encoding="utf-8", + ) + else: + with pytest.raises(ValueError, match=error): + store.append_snapshot_delta(str(path), sequence=1, replace_from=0, messages=[message], meta={}) + assert not store.has_snapshot(str(path)) + return + with pytest.raises(ValueError, match=error): + store.load_snapshot(str(path), "fallback")