mirror of
https://github.com/NCBM/plyngent.git
synced 2026-07-23 05:55:16 +08:00
core/agent+cli: fix compact session restore and persist cursor
After /compact, reload seed from DB so _persist_from matches stored rows. rebuild_client preserves the checkpoint via replace_messages. Clarify that the active session is summary-only and full history stays on the old id.
This commit is contained in:
@@ -200,14 +200,38 @@ class ChatAgent:
|
|||||||
if self._persist_from == 0:
|
if self._persist_from == 0:
|
||||||
self._persist_from = 1
|
self._persist_from = 1
|
||||||
|
|
||||||
|
def replace_messages(
|
||||||
|
self,
|
||||||
|
messages: Sequence[AnyChatMessage],
|
||||||
|
*,
|
||||||
|
persisted: bool = True,
|
||||||
|
persist_from: int | None = None,
|
||||||
|
) -> None:
|
||||||
|
"""Replace in-memory history and align the persistence checkpoint.
|
||||||
|
|
||||||
|
When *persist_from* is set, it is used as the checkpoint cursor
|
||||||
|
(clamped to ``[0, len(messages)]``). Otherwise *persisted* true means
|
||||||
|
all messages are already stored; false means none are.
|
||||||
|
"""
|
||||||
|
self.messages = list(messages)
|
||||||
|
if persist_from is not None:
|
||||||
|
self._persist_from = max(0, min(persist_from, len(self.messages)))
|
||||||
|
else:
|
||||||
|
self._persist_from = len(self.messages) if persisted else 0
|
||||||
|
self._ensure_system_prompt()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def persist_from(self) -> int:
|
||||||
|
"""Index of the first unpersisted message (checkpoint cursor)."""
|
||||||
|
return self._persist_from
|
||||||
|
|
||||||
async def load_history(self) -> None:
|
async def load_history(self) -> None:
|
||||||
"""Replace in-memory messages from the bound memory session."""
|
"""Replace in-memory messages from the bound memory session."""
|
||||||
if self.memory is None or self.session_id is None:
|
if self.memory is None or self.session_id is None:
|
||||||
msg = "load_history requires memory and session_id"
|
msg = "load_history requires memory and session_id"
|
||||||
raise RuntimeError(msg)
|
raise RuntimeError(msg)
|
||||||
self.messages = await self.memory.list_messages(self.session_id)
|
loaded = await self.memory.list_messages(self.session_id)
|
||||||
self._persist_from = len(self.messages)
|
self.replace_messages(loaded, persisted=True)
|
||||||
self._ensure_system_prompt()
|
|
||||||
|
|
||||||
async def bind_session(self, session_id: int, *, load: bool = True) -> None:
|
async def bind_session(self, session_id: int, *, load: bool = True) -> None:
|
||||||
"""Attach a memory session id; optionally load existing messages."""
|
"""Attach a memory session id; optionally load existing messages."""
|
||||||
|
|||||||
@@ -576,6 +576,11 @@ def compact_cmd(state: ReplState, name: str | None) -> None:
|
|||||||
return
|
return
|
||||||
preview = summary if len(summary) <= _COMPACT_PREVIEW else summary[:_COMPACT_PREVIEW] + "…"
|
preview = summary if len(summary) <= _COMPACT_PREVIEW else summary[:_COMPACT_PREVIEW] + "…"
|
||||||
click.echo(f"compacted session {old_id} -> new session {new_id}")
|
click.echo(f"compacted session {old_id} -> new session {new_id}")
|
||||||
|
click.secho(
|
||||||
|
f"active session is summary-only ({len(state.agent.messages)} messages); "
|
||||||
|
f"full history remains on /resume {old_id}",
|
||||||
|
fg="bright_black",
|
||||||
|
)
|
||||||
click.secho(preview, fg="bright_black")
|
click.secho(preview, fg="bright_black")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -145,12 +145,14 @@ class ReplState:
|
|||||||
def rebuild_client(self) -> None:
|
def rebuild_client(self) -> None:
|
||||||
"""Recreate client and agent after provider/model/tools change."""
|
"""Recreate client and agent after provider/model/tools change."""
|
||||||
messages = list(self.agent.messages)
|
messages = list(self.agent.messages)
|
||||||
|
persist_from = self.agent.persist_from
|
||||||
# Preserve live stream toggle if agent already exists.
|
# Preserve live stream toggle if agent already exists.
|
||||||
if hasattr(self, "agent"):
|
if hasattr(self, "agent"):
|
||||||
self.stream_enabled = self.agent.stream
|
self.stream_enabled = self.agent.stream
|
||||||
self.client = cast("ChatClient", cast("object", create_client(self.provider)))
|
self.client = cast("ChatClient", cast("object", create_client(self.provider)))
|
||||||
self.agent = self._make_agent()
|
self.agent = self._make_agent()
|
||||||
self.agent.messages = messages
|
# Restore history without re-marking already-stored messages as dirty.
|
||||||
|
self.agent.replace_messages(messages, persist_from=persist_from)
|
||||||
self.sync_display_flags()
|
self.sync_display_flags()
|
||||||
# Drop remote catalog when provider identity/url changed (not on model-only switch).
|
# Drop remote catalog when provider identity/url changed (not on model-only switch).
|
||||||
if self._remote_models_key is not None and self._remote_models_key != self._models_cache_key():
|
if self._remote_models_key is not None and self._remote_models_key != self._models_cache_key():
|
||||||
@@ -474,7 +476,12 @@ class ReplState:
|
|||||||
source_session_id=old_id,
|
source_session_id=old_id,
|
||||||
seed_text=self.config.agent_config.compact_seed_text or None,
|
seed_text=self.config.agent_config.compact_seed_text or None,
|
||||||
)
|
)
|
||||||
self.agent.messages = list(seed)
|
|
||||||
for message in seed:
|
for message in seed:
|
||||||
_ = await self.memory.append_message(new_id, message)
|
_ = await self.memory.append_message(new_id, message)
|
||||||
|
# Reload from DB so RAM matches stored rows and the persist cursor is correct
|
||||||
|
# (assigning messages alone left _persist_from at 0 and broke later checkpoints).
|
||||||
|
await self.agent.load_history()
|
||||||
|
if not self.agent.messages:
|
||||||
|
msg = f"compact session {new_id} has no messages after seed"
|
||||||
|
raise RuntimeError(msg)
|
||||||
return old_id, new_id, summary
|
return old_id, new_id, summary
|
||||||
|
|||||||
@@ -99,11 +99,63 @@ async def test_compact_to_new_session(tmp_path: Path) -> None:
|
|||||||
assert "compacted goals" in summary
|
assert "compacted goals" in summary
|
||||||
loaded = await memory.list_messages(new_id)
|
loaded = await memory.list_messages(new_id)
|
||||||
assert any("compacted goals" in getattr(m, "content", "") for m in loaded)
|
assert any("compacted goals" in getattr(m, "content", "") for m in loaded)
|
||||||
|
# After compact, RAM matches DB and the persist cursor treats seed as stored.
|
||||||
|
assert len(state.agent.messages) == len(loaded)
|
||||||
|
assert state.agent.persist_from == len(state.agent.messages)
|
||||||
# Old session still exists and is listable
|
# Old session still exists and is listable
|
||||||
sessions = await memory.list_sessions(workspace=tmp_path)
|
sessions = await memory.list_sessions(workspace=tmp_path)
|
||||||
ids = {s.sid for s in sessions}
|
ids = {s.sid for s in sessions}
|
||||||
assert old_id in ids
|
assert old_id in ids
|
||||||
assert new_id in ids
|
assert new_id in ids
|
||||||
|
|
||||||
|
# Simulate restart: resume compact session, continue without duplicating seed.
|
||||||
|
state2 = ReplState(
|
||||||
|
config=config,
|
||||||
|
memory=memory,
|
||||||
|
workspace=tmp_path,
|
||||||
|
provider_name="local",
|
||||||
|
provider=provider,
|
||||||
|
model="gpt-test",
|
||||||
|
tools_enabled=False,
|
||||||
|
)
|
||||||
|
state2.client = SummaryClient()
|
||||||
|
await state2.resume_session(new_id)
|
||||||
|
assert len(state2.agent.messages) == len(loaded)
|
||||||
|
assert state2.agent.persist_from == len(state2.agent.messages)
|
||||||
|
assert any("compacted goals" in getattr(m, "content", "") for m in state2.agent.messages)
|
||||||
|
# Full original conversation still only on the old session.
|
||||||
|
old_loaded = await memory.list_messages(old_id)
|
||||||
|
assert any(isinstance(m, UserChatMessage) and m.content == "do work" for m in old_loaded)
|
||||||
|
finally:
|
||||||
|
await memory.close()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_rebuild_client_preserves_persist_cursor(tmp_path: Path) -> None:
|
||||||
|
_ = set_workspace_root(tmp_path)
|
||||||
|
memory = await MemoryStore.open(DatabaseConfig())
|
||||||
|
try:
|
||||||
|
provider = OpenAIProvider(access_key_or_token="sk-test")
|
||||||
|
config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
|
||||||
|
config.providers = {"local": provider}
|
||||||
|
state = ReplState(
|
||||||
|
config=config,
|
||||||
|
memory=memory,
|
||||||
|
workspace=tmp_path,
|
||||||
|
provider_name="local",
|
||||||
|
provider=provider,
|
||||||
|
model="gpt-test",
|
||||||
|
tools_enabled=False,
|
||||||
|
)
|
||||||
|
state.client = SummaryClient()
|
||||||
|
await state.new_session("s")
|
||||||
|
state.agent.replace_messages(
|
||||||
|
[UserChatMessage(content="hi"), AssistantChatMessage(content="yo")],
|
||||||
|
persisted=True,
|
||||||
|
)
|
||||||
|
assert state.agent.persist_from == 2
|
||||||
|
state.rebuild_client()
|
||||||
|
assert len(state.agent.messages) == 2
|
||||||
|
assert state.agent.persist_from == 2
|
||||||
finally:
|
finally:
|
||||||
await memory.close()
|
await memory.close()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user