让 Agent 记住上一次对话 — REPL 会话管理的工程实现

_SESSION_ID_TIMESTAMP_FORMAT = "%Y%m%d-%H%M%S-%f"

def _generate_session_id(
    transcript_mgr: TranscriptManager,
    *,
    reserved_ids: set[str] | None = None,
) -> str:
    known_session_ids = set(transcript_mgr.list_sessions())
    if reserved_ids:
        known_session_ids.update(
            session_id for session_id in reserved_ids if session_id
        )

    while True:
        candidate = (
            f"{datetime.now().strftime(_SESSION_ID_TIMESTAMP_FORMAT)}"
            f"-{generate_random_id(6)}"
        )
        if candidate not in known_session_ids:
            return candidate

class TranscriptManager:
    def __init__(self, transcript_dir: str | Path = ".transcripts") -> None:
        self.transcript_dir = Path(transcript_dir)
        self.transcript_dir.mkdir(parents=True, exist_ok=True)

    def save(self, messages: list[dict[str, Any]], session_id: str) -> Path:
        timestamp = datetime.now().strftime(_SAVE_TIMESTAMP_FORMAT)
        path = self.transcript_dir / f"{session_id}_{timestamp}.jsonl"
        with path.open("w", encoding="utf-8") as file:
            for message in messages:
                file.write(json.dumps(message, ensure_ascii=False))
                file.write("\n")
        return path

    def load(self, session_id: str) -> list[dict[str, Any]]:
        entry = self._get_session_entry(session_id)
        with entry.path.open("r", encoding="utf-8") as file:
            return [json.loads(line) for line in file if line.strip()]

def list_sessions(self) -> list[str]:
    latest_by_session: dict[str, datetime] = {}
    for entry in self._iter_entries():
        latest_by_session[entry.session_id] = max(
            entry.timestamp,
            latest_by_session.get(entry.session_id, datetime.min),
        )
    return [
        session_id
        for session_id, _ in sorted(
            latest_by_session.items(), key=lambda item: item[1], reverse=True,
        )
    ]

def resume(self, session_id: str | None = None) -> list[dict[str, Any]]:
    target_session = session_id or self.get_latest_session()
    if target_session is None:
        raise FileNotFoundError("No saved transcripts found.")
    return self.load(target_session)

if text == "/resume" or text.startswith("/resume "):
    _, _, raw_session_id = text.partition(" ")
    requested_session = raw_session_id.strip() or None
    try:
        restored_messages = transcript_mgr.resume(requested_session)
    except FileNotFoundError as exc:
        ui_console.print_error(str(exc))
        continue
    messages[:] = restored_messages
    resumed_session = requested_session or transcript_mgr.get_latest_session()
    if resumed_session is not None:
        _set_compact_session_id(compact_fn, resumed_session)
        _set_interaction_logger_session(interaction_logger, resumed_session)
        message_bus, main_mailbox_cursor = _switch_session_mailbox(
            workspace_path, resumed_session, current_bus=message_bus,
        )
        spawned_agents = {}
    handlers = _build_handlers(
        workspace_path=workspace_path,
        todo_manager=todo_manager,
        # ... 省略其他参数
        messages=messages,
        message_bus=message_bus,
        spawned_agents=spawned_agents,
    )
    ui_console.print_status(f"Resumed session: {resumed_session}")
    continue

def _refresh_nag_reminder(messages, nag_reminder):
    # 先删掉所有旧的 nag reminder
    messages[:] = [
        msg for msg in messages
        if not (msg.get("role") == "system"
                and str(msg["content"]).startswith(_NAG_REMINDER_PREFIX))
    ]
    if not nag_reminder:
        return

    nag_message = {
        "role": "system",
        "content": f"{_NAG_REMINDER_PREFIX}\n{nag_reminder.strip()}\n</nag-reminder>",
    }
    # 插入到最后一条 user message 之后
    for index in range(len(messages) - 1, -1, -1):
        if messages[index].get("role") == "user" and not is_tool_result_message(messages[index]):
            messages.insert(index + 1, nag_message)
            return
    messages.append(nag_message)

messages.append({"role": "user", "content": text})
snapshot_len = len(messages) - 1
try:
    agent_loop(
        provider=provider, messages=messages,
        tools=tools, handlers=handlers,
        # ...
    )
    _save_transcript_snapshot(transcript_mgr, messages, compact_fn)
except LLMCallError:
    del messages[snapshot_len:]
    ui_console.print_error("LLM call failed, please try again.")
except KeyboardInterrupt:
    del messages[snapshot_len:]
    ui_console.print_status("Agent loop interrupted.")

def _switch_session_mailbox(workspace_path, session_id, *, current_bus=None):
    if current_bus is not None:
        _broadcast_team_shutdown(current_bus)

    message_bus = MessageBus(workspace_path / ".mailbox" / session_id)
    message_bus.ensure_mailbox(MAIN_AGENT_NAME)
    return message_bus, message_bus.latest_message_id(MAIN_AGENT_NAME)