From 5d8a43655127c9f100b537e02f8c0dc02dc7dbf8 Mon Sep 17 00:00:00 2001 From: Xiaoyu <458788361@qq.com> Date: Thu, 25 Jun 2026 11:40:55 +0800 Subject: [PATCH 1/2] feat(intake): persist and resume the planning conversation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The intake REPL (`arbor` with no subcommand) is a planning chat held before a research run is launched. Until now it lived only in memory: quit before launching and it was gone, so there was nothing to resume — only launched runs under .arbor/sessions/ were ever persisted. Auto-save the conversation every turn under /.arbor/conversations// (messages.jsonl + meta.json), reusing the same atomic JSONL message IO as run checkpoints. Add two ways back in: - `arbor --continue` / `-C`: reload the newest unfinished conversation and keep chatting (like `claude -c`). `-c` is already `--config`, so the short flag is `-C`. - `/resume` inside intake now lists saved conversations alongside launched runs; picking a conversation reloads it into the live agent, picking a run replays the orchestrator as before. On resume (both paths) the prior exchange is replayed to the terminal so you can see the previous conversation, not just silently inherit its context. A conversation is marked `launched` once it fires LaunchExperiment (the run's own checkpoint takes over) and is then excluded from --continue. Saves are best-effort: a failure logs and never breaks the chat. New module src/cli/intake/conversation_store.py with full unit tests, plus integration tests driving run_intake with the LLM call and input stubbed. --- src/cli/commands/run.py | 7 + src/cli/intake/conversation_store.py | 240 ++++++++++++++++++++++++ src/cli/intake/launch_tool.py | 4 + src/cli/intake/repl.py | 175 +++++++++++++++-- tests/test_intake_conversation_store.py | 128 +++++++++++++ tests/test_intake_repl_autosave.py | 139 ++++++++++++++ 6 files changed, 676 insertions(+), 17 deletions(-) create mode 100644 src/cli/intake/conversation_store.py create mode 100644 tests/test_intake_conversation_store.py create mode 100644 tests/test_intake_repl_autosave.py diff --git a/src/cli/commands/run.py b/src/cli/commands/run.py index de5ba82..8834b16 100644 --- a/src/cli/commands/run.py +++ b/src/cli/commands/run.py @@ -51,6 +51,12 @@ def run_command( help="Resume an interrupted run from its checkpoint (idea tree + message history) " "in the existing workspace/session, instead of starting fresh.", ), + continue_latest: bool = typer.Option( + False, "--continue", "-C", + help="Continue the most recent unfinished intake conversation in this directory " + "(like `claude -c`). Reloads the planning chat so you pick up where you left off. " + "(-c is --config; use -C or --continue.)", + ), workspace_dir: Path | None = typer.Option( None, "--workspace-dir", help=f"Session/artifact directory override. Default: /{CONFIG_DIR_NAME}/sessions/.", @@ -243,6 +249,7 @@ def _probe_config() -> CoordinatorConfig: starting_cwd=starting_cwd, seed_message=instruction, intake_max_turns=intake_max_turns, + continue_latest=continue_latest, )) except KeyboardInterrupt: typer.secho("\n^C — aborted before run started", fg=typer.colors.YELLOW, err=True) diff --git a/src/cli/intake/conversation_store.py b/src/cli/intake/conversation_store.py new file mode 100644 index 0000000..9ac4e91 --- /dev/null +++ b/src/cli/intake/conversation_store.py @@ -0,0 +1,240 @@ +"""Persist the pre-launch intake conversation so it can be resumed. + +The intake REPL (``arbor`` with no subcommand) is a planning chat held *before* +a research run is launched. Until now that chat lived only in memory: quit +before launching and it was gone, so there was nothing to ``/resume``. Launched +runs persist under ``.arbor/sessions//`` (see +:mod:`arbor.coordinator.checkpoint`); this module is the analogous, deliberately +lighter store for the *conversation* itself. + +Layout, one dir per conversation under the launch directory:: + + /.arbor/conversations// + messages.jsonl # the agent's message history (atomic JSONL IO) + meta.json # id, timestamps, title, turn count, launched flag + +A conversation is **unfinished** while ``launched`` is false. Once the chat +fires ``LaunchExperiment`` the run's own checkpoint takes over, so the record is +marked ``launched`` and excluded from ``--continue`` (it is kept as history). + +Message IO is reused verbatim from the checkpoint contract +(:func:`arbor.coordinator.checkpoint.write_messages` / +:func:`~arbor.coordinator.checkpoint.read_messages`) so a conversation and a run +serialize their histories identically. Meta writes are atomic (temp + replace) +for the same reason checkpoints are: an interrupt must never leave a torn file. +""" + +from __future__ import annotations + +import json +import os +import tempfile +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from ..._app import CONFIG_DIR_NAME +from ...coordinator.checkpoint import read_messages, write_messages + +#: Bump when the on-disk meta shape changes incompatibly. +SCHEMA_VERSION = 1 + +CONVERSATIONS_DIRNAME = "conversations" +MESSAGES_NAME = "messages.jsonl" +META_NAME = "meta.json" + +_TITLE_MAX = 80 + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def conversations_root(cwd: str | os.PathLike[str]) -> Path: + """Return ``/.arbor/conversations`` (the per-project conversation root).""" + return Path(cwd) / CONFIG_DIR_NAME / CONVERSATIONS_DIRNAME + + +@dataclass +class ConversationRecord: + """One saved intake conversation, addressed by ``conv_id`` under ``cwd``.""" + + conv_id: str + cwd: Path + created_at: str + updated_at: str + title: str = "" + turns: int = 0 + launched: bool = False + + @property + def dir(self) -> Path: + return conversations_root(self.cwd) / self.conv_id + + @property + def messages_path(self) -> Path: + return self.dir / MESSAGES_NAME + + @property + def meta_path(self) -> Path: + return self.dir / META_NAME + + def to_meta(self) -> dict[str, Any]: + return { + "schema_version": SCHEMA_VERSION, + "conv_id": self.conv_id, + "created_at": self.created_at, + "updated_at": self.updated_at, + "title": self.title, + "turns": self.turns, + "launched": self.launched, + } + + @classmethod + def from_meta(cls, cwd: Path, data: dict[str, Any]) -> "ConversationRecord": + return cls( + conv_id=str(data["conv_id"]), + cwd=Path(cwd), + created_at=str(data.get("created_at") or ""), + updated_at=str(data.get("updated_at") or data.get("created_at") or ""), + title=str(data.get("title") or ""), + turns=int(data.get("turns") or 0), + launched=bool(data.get("launched", False)), + ) + + +def new_conversation(cwd: str | os.PathLike[str]) -> ConversationRecord: + """Mint a fresh record (timestamp-derived id). Does NOT touch disk. + + The id is unique even within the same second: if a directory already + exists, a numeric suffix is appended. + """ + now = datetime.now() + base = now.strftime("conv_%Y%m%d_%H%M%S") + root = conversations_root(cwd) + conv_id = base + n = 1 + while (root / conv_id).exists(): + n += 1 + conv_id = f"{base}_{n}" + iso = _utc_now_iso() + return ConversationRecord( + conv_id=conv_id, cwd=Path(cwd), created_at=iso, updated_at=iso + ) + + +def save_conversation( + rec: ConversationRecord, + messages: list[dict[str, Any]], + *, + launched: bool = False, +) -> None: + """Persist ``messages`` + refreshed meta for ``rec`` atomically. + + Updates ``rec`` in place (``updated_at``, ``turns``, ``title``, ``launched``) + so the caller's handle stays current across repeated saves. + """ + rec.dir.mkdir(parents=True, exist_ok=True) + write_messages(rec.messages_path, messages) + + rec.updated_at = _utc_now_iso() + rec.turns = _count_user_turns(messages) + rec.launched = launched + if not rec.title: + rec.title = _derive_title(messages) + + _atomic_write_json(rec.meta_path, rec.to_meta()) + + +def load_messages(rec: ConversationRecord) -> list[dict[str, Any]]: + """Load the saved message history (``[]`` if none), tolerant of corruption.""" + return read_messages(rec.messages_path) + + +def find_conversations(cwd: str | os.PathLike[str]) -> list[ConversationRecord]: + """Return all readable conversations under ``cwd``, newest update first. + + Defensive: a dir without a parseable ``meta.json`` is skipped, never fatal. + """ + root = conversations_root(cwd) + if not root.is_dir(): + return [] + + records: list[ConversationRecord] = [] + for conv_dir in root.iterdir(): + if not conv_dir.is_dir(): + continue + data = _load_json(conv_dir / META_NAME) + if not isinstance(data, dict) or "conv_id" not in data: + continue + try: + records.append(ConversationRecord.from_meta(Path(cwd), data)) + except (KeyError, ValueError, TypeError): + continue + + records.sort(key=lambda r: r.updated_at, reverse=True) + return records + + +def latest_unfinished(cwd: str | os.PathLike[str]) -> ConversationRecord | None: + """Return the newest non-launched conversation with real content, or None.""" + for rec in find_conversations(cwd): + if not rec.launched and rec.turns > 0 and rec.messages_path.is_file(): + return rec + return None + + +# ── helpers ────────────────────────────────────────────────────────── + + +def _count_user_turns(messages: list[dict[str, Any]]) -> int: + return sum(1 for m in messages if m.get("role") == "user") + + +def _derive_title(messages: list[dict[str, Any]]) -> str: + """First user message, flattened to a short single line.""" + for m in messages: + if m.get("role") != "user": + continue + text = _message_text(m.get("content")) + text = " ".join(text.split()) + if text: + return text if len(text) <= _TITLE_MAX else text[: _TITLE_MAX - 1] + "…" + return "" + + +def _message_text(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [ + b.get("text", "") + for b in content + if isinstance(b, dict) and isinstance(b.get("text"), str) + ] + return " ".join(p for p in parts if p) + return "" + + +def _atomic_write_json(path: Path, data: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = json.dumps(data, indent=2, ensure_ascii=False) + fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp") + tmp = Path(tmp_name) + try: + with os.fdopen(fd, "w", encoding="utf-8") as fp: + fp.write(payload) + fp.flush() + os.fsync(fp.fileno()) + os.replace(tmp, path) + except BaseException: + tmp.unlink(missing_ok=True) + raise + + +def _load_json(path: Path) -> Any | None: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None diff --git a/src/cli/intake/launch_tool.py b/src/cli/intake/launch_tool.py index 7098e4a..de83e84 100644 --- a/src/cli/intake/launch_tool.py +++ b/src/cli/intake/launch_tool.py @@ -46,6 +46,10 @@ class LaunchState: # Set when the user runs `/resume` and picks a past session. Typed Any to # avoid importing resume_picker here; it holds a ``ResumableSession``. resume_target: Any = None + # Set when `/resume` picks a saved *conversation* (not a launched run): the + # chosen ``ConversationRecord``. The REPL reloads it into the live agent and + # keeps chatting, persisting onward saves to this same record. + resume_conversation: Any = None @property def launched(self) -> bool: diff --git a/src/cli/intake/repl.py b/src/cli/intake/repl.py index 75114e4..06eec10 100644 --- a/src/cli/intake/repl.py +++ b/src/cli/intake/repl.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from pathlib import Path from typing import Callable, Iterable @@ -30,6 +31,15 @@ from ...core.tools.file_read import FileReadTool from ...core.tools.glob_tool import GlobTool from ...core.tools.grep import GrepTool +from ...coordinator.checkpoint import seal_interrupted_tail +from .conversation_store import ( + ConversationRecord, + find_conversations, + latest_unfinished, + load_messages, + new_conversation, + save_conversation, +) from .display import IntakeDisplay from .launch_tool import LaunchExperimentTool, LaunchPlan, LaunchState from .system_prompt import build_system_prompt @@ -37,13 +47,14 @@ _console = Console() +log = logging.getLogger(__name__) # Single source of truth for slash commands. # (name, description) — the completer renders these, _handle_slash dispatches. SLASH_COMMANDS: list[tuple[str, str]] = [ ("/help", "show available commands"), - ("/resume", "resume a previous run"), + ("/resume", "resume a past conversation or run"), ("/plugin", "select plugin: /plugin [profile]"), ("/skill", "manage skills: /skill load|unload "), ("/status", "show intake status"), @@ -63,6 +74,7 @@ async def run_intake( seed_message: str | None, workspace_dir: Path | None = None, intake_max_turns: int = 30, + continue_latest: bool = False, ) -> LaunchPlan | ResumableSession | None: """Drive the intake REPL. Returns the launch outcome: @@ -73,6 +85,11 @@ async def run_intake( `starting_cwd` is the directory the user invoked the CLI from. It is only a hint — the agent must confirm or correct it before launching. The intake workspace dir (tool persistence, etc.) lives next to it. + + The conversation itself is auto-saved every turn under + ``/.arbor/conversations/`` so it can be continued later. With + ``continue_latest`` (``arbor --continue``) the newest unfinished + conversation there is reloaded and the chat picks up where it left off. """ starting_cwd = starting_cwd.resolve() @@ -117,6 +134,33 @@ async def run_intake( session = _build_session(starting_cwd, state=state) + # The conversation is auto-saved every turn so it can be resumed. On + # --continue, reload the newest unfinished conversation and keep chatting; + # otherwise start a fresh record (written lazily on the first save). + conv: ConversationRecord = new_conversation(starting_cwd) + if continue_latest: + prior = latest_unfinished(starting_cwd) + if prior is not None: + agent.messages = seal_interrupted_tail(load_messages(prior)) + conv = prior + _console.print( + f"[green]Continuing your last conversation[/] " + f"[dim]({escape(prior.title) or prior.conv_id})[/]\n" + ) + _print_resumed_history(agent.messages) + else: + _console.print( + "[yellow]No unfinished conversation to continue[/] " + "[dim]— starting fresh.[/]\n" + ) + + def _persist() -> None: + """Best-effort autosave; a failure must never break the chat.""" + try: + save_conversation(conv, agent.messages, launched=state.launched) + except Exception as exc: # pylint: disable=broad-exception-caught + log.debug("intake conversation save failed: %s", exc) + # Seed message: if the user gave one on the CLI, feed it as the first user turn pending_user_input: str | None = seed_message @@ -152,6 +196,12 @@ async def run_intake( # The user picked a past run via /resume — hand it straight back # so run_command resumes it instead of starting a fresh run. return state.resume_target + if action == "conversation": + # The user picked a saved conversation: _handle_slash already + # reloaded it into the agent. Switch onward autosaves to that + # record and keep chatting in this same intake loop. + conv = state.resume_conversation + continue continue # Hand off to the agent under a compact display. @@ -171,6 +221,10 @@ async def run_intake( ) continue + # Autosave the turn (records launched=True once a plan is fired, so a + # launched conversation is excluded from --continue). + _persist() + # The agent prints its own messages via core/agent.py's _print_* # helpers, and it only fires LaunchExperiment after the user has # explicitly approved the plan in conversation. So we hand the plan @@ -520,6 +574,85 @@ def _print_research_contract(plan: LaunchPlan) -> None: _console.print() +def _visible_text(content: object) -> str: + """Human-readable text of a message's content, ignoring tool plumbing.""" + if isinstance(content, str): + return content.strip() + if isinstance(content, list): + parts = [ + b.get("text", "") + for b in content + if isinstance(b, dict) and b.get("type") in (None, "text") and isinstance(b.get("text"), str) + ] + return " ".join(p for p in parts if p).strip() + return "" + + +def _print_resumed_history(messages: list[dict]) -> None: + """Replay a reloaded conversation so the user can see the prior exchange. + + Renders user prompts and the agent's text replies; tool calls / results + carry no prose and are skipped. No-op for an empty history. + """ + rendered = False + for m in messages: + text = _visible_text(m.get("content")) + if not text: + continue + if not rendered: + _console.print("[dim]──────── previous conversation ────────[/dim]") + rendered = True + role = m.get("role") + if role == "user": + _console.print(f"[dim]you ›[/dim] {escape(text)}") + elif role == "assistant": + _console.print(f"[cyan]arbor ›[/cyan] {escape(text)}") + if rendered: + _console.print("[dim]───────────────────────────────────────[/dim]\n") + + +def _prompt_resume_any( + convs: list[ConversationRecord], + runs: list[ResumableSession], + *, + console: Console, +) -> ConversationRecord | ResumableSession | None: + """Show conversations then runs as one numbered list; return the choice. + + Returns the picked :class:`ConversationRecord` or :class:`ResumableSession`, + or ``None`` when the user starts fresh. + """ + from ..resume_picker import _format_row, _humanize_age, _parse_iso + + items: list[ConversationRecord | ResumableSession] = [*convs, *runs] + console.print() + if convs: + console.print("[bold cyan]Conversations[/] [dim](continue chatting)[/]") + for i, c in enumerate(convs, 1): + age = _humanize_age(_parse_iso(c.updated_at)) + title = escape(c.title or "(no messages yet)") + console.print( + f" [bold]{i}[/] [cyan]{escape(c.conv_id)}[/] [dim]{age}[/] " + f"[dim]{c.turns} turn(s)[/] [magenta]\\[chat][/]\n" + f" [dim]{title}[/]" + ) + if runs: + console.print("[bold cyan]Runs[/] [dim](replay the research engine)[/]") + for i, s in enumerate(runs, len(convs) + 1): + console.print(_format_row(i, s)) + console.print() + + while True: + answer = typer.prompt( + f"Start fresh, or resume one? [N / 1-{len(items)}]", default="N" + ).strip().lower() + if answer in ("", "n", "new"): + return None + if answer.isdigit() and 1 <= int(answer) <= len(items): + return items[int(answer) - 1] + console.print(f"[yellow] enter N for a new chat, or 1-{len(items)} to resume[/]") + + def _handle_slash( line: str, agent: Agent, @@ -535,32 +668,40 @@ def _handle_slash( for name, desc in SLASH_COMMANDS: _console.print(f" [cyan]{name:<8}[/cyan] [dim]{desc}[/dim]") elif cmd == "/resume": - # List this project's past runs and let the user pick one to resume. - # Reuses the standalone picker; a chosen session is stashed on the - # shared state and the main loop returns it from run_intake. - from ..resume_picker import find_resumable_sessions, prompt_resume_choice + # List this project's past *conversations* and launched *runs* in one + # picker. A conversation is reloaded into the live agent (keep chatting); + # a run is handed back so run_command replays the orchestrator. + from ..resume_picker import find_resumable_sessions - sessions = ( + convs = ( + [c for c in find_conversations(starting_cwd) if not c.launched and c.turns > 0] + if starting_cwd else [] + ) + runs = ( find_resumable_sessions(starting_cwd, include_subdirs=True) if starting_cwd else [] ) - if not sessions: + if not convs and not runs: where = escape(str(starting_cwd)) if starting_cwd else "this directory" + _console.print(f"[yellow]Nothing to resume[/] under [dim]{where}[/].") _console.print( - f"[yellow]No resumable runs found[/] under [dim]{where}[/]." - ) - _console.print( - " [dim]Resume is per-project: runs live in " - "[/dim][cyan].arbor/sessions/[/cyan][dim] under the project you " - "worked on.\n" - " Launch [/dim][cyan]arbor[/cyan][dim] from inside (or just " - "above) that project, or start a new run by describing your " - "goal.[/dim]" + " [dim]Conversations live in [/dim][cyan].arbor/conversations/[/cyan]" + "[dim] and launched runs in [/dim][cyan].arbor/sessions/[/cyan][dim], " + "per project. Start a new one by describing your goal.[/dim]" ) return "continue" - chosen = prompt_resume_choice(sessions, console=_console) + chosen = _prompt_resume_any(convs, runs, console=_console) if chosen is None: return "continue" # user declined (N) → stay in intake + if isinstance(chosen, ConversationRecord): + agent.messages = seal_interrupted_tail(load_messages(chosen)) + state.resume_conversation = chosen + _console.print( + f"[green]Resumed conversation[/] " + f"[dim]({escape(chosen.title) or chosen.conv_id})[/]\n" + ) + _print_resumed_history(agent.messages) + return "conversation" state.resume_target = chosen return "resume" elif cmd == "/status": diff --git a/tests/test_intake_conversation_store.py b/tests/test_intake_conversation_store.py new file mode 100644 index 0000000..6332cf1 --- /dev/null +++ b/tests/test_intake_conversation_store.py @@ -0,0 +1,128 @@ +"""Unit tests for the intake conversation store. + +The store persists the *pre-launch intake conversation* (the chat you have with +``arbor`` before a research run is launched) so it can be auto-continued +(``arbor --continue``) or picked from ``/resume``. It reuses the same atomic +JSONL message IO as run checkpoints. +""" + +from __future__ import annotations + +import json + +from arbor.cli.intake.conversation_store import ( + ConversationRecord, + conversations_root, + find_conversations, + latest_unfinished, + load_messages, + new_conversation, + save_conversation, +) + + +def _msgs(*user_texts: str) -> list[dict]: + out: list[dict] = [] + for t in user_texts: + out.append({"role": "user", "content": t}) + out.append({"role": "assistant", "content": [{"type": "text", "text": "ok"}]}) + return out + + +def test_new_conversation_is_not_written_until_saved(tmp_path): + rec = new_conversation(tmp_path) + assert isinstance(rec, ConversationRecord) + assert rec.conv_id.startswith("conv_") + assert rec.cwd == tmp_path + # Nothing on disk yet. + assert not rec.dir.exists() + assert find_conversations(tmp_path) == [] + + +def test_save_then_load_round_trips_messages(tmp_path): + rec = new_conversation(tmp_path) + messages = _msgs("first task", "second task") + save_conversation(rec, messages) + + assert rec.messages_path.is_file() + assert rec.meta_path.is_file() + assert load_messages(rec) == messages + + +def test_meta_records_title_turns_and_launched(tmp_path): + rec = new_conversation(tmp_path) + save_conversation(rec, _msgs("optimize the dev score please"), launched=False) + + meta = json.loads(rec.meta_path.read_text(encoding="utf-8")) + assert meta["title"].startswith("optimize the dev score") + assert meta["turns"] == 1 # one user message + assert meta["launched"] is False + assert meta["conv_id"] == rec.conv_id + + +def test_find_conversations_newest_first(tmp_path): + a = new_conversation(tmp_path) + save_conversation(a, _msgs("alpha")) + b = new_conversation(tmp_path) + save_conversation(b, _msgs("beta")) + # Force a strictly newer update on `a`. + save_conversation(a, _msgs("alpha", "alpha again")) + + found = find_conversations(tmp_path) + assert [r.conv_id for r in found][0] == a.conv_id + assert {r.conv_id for r in found} == {a.conv_id, b.conv_id} + + +def test_latest_unfinished_skips_launched_and_empty(tmp_path): + # Launched conversation — excluded. + launched = new_conversation(tmp_path) + save_conversation(launched, _msgs("this one launched"), launched=True) + + # Unfinished with content — the expected pick. + unfinished = new_conversation(tmp_path) + save_conversation(unfinished, _msgs("still planning"), launched=False) + + pick = latest_unfinished(tmp_path) + assert pick is not None + assert pick.conv_id == unfinished.conv_id + + +def test_latest_unfinished_none_when_all_launched(tmp_path): + rec = new_conversation(tmp_path) + save_conversation(rec, _msgs("done"), launched=True) + assert latest_unfinished(tmp_path) is None + + +def test_latest_unfinished_none_on_empty_dir(tmp_path): + assert latest_unfinished(tmp_path) is None + assert conversations_root(tmp_path) == tmp_path / ".arbor" / "conversations" + + +def test_save_is_atomic_and_meta_is_valid_json(tmp_path): + rec = new_conversation(tmp_path) + save_conversation(rec, _msgs("hello")) + # No leftover temp files in the conversation dir. + leftovers = [p.name for p in rec.dir.iterdir() if p.name.startswith(".")] + assert leftovers == [] + json.loads(rec.meta_path.read_text(encoding="utf-8")) # must parse + + +def test_corrupt_meta_is_skipped_not_fatal(tmp_path): + good = new_conversation(tmp_path) + save_conversation(good, _msgs("good one")) + + bad_dir = conversations_root(tmp_path) / "conv_broken" + bad_dir.mkdir(parents=True) + (bad_dir / "meta.json").write_text("{ not json", encoding="utf-8") + + found = find_conversations(tmp_path) + assert {r.conv_id for r in found} == {good.conv_id} # broken one skipped + + +def test_new_conversation_ids_are_unique(tmp_path): + ids = set() + for _ in range(5): + rec = new_conversation(tmp_path) + save_conversation(rec, _msgs("x")) + ids.add(rec.conv_id) + assert len(ids) == 5 diff --git a/tests/test_intake_repl_autosave.py b/tests/test_intake_repl_autosave.py new file mode 100644 index 0000000..21bf990 --- /dev/null +++ b/tests/test_intake_repl_autosave.py @@ -0,0 +1,139 @@ +"""Integration tests for intake conversation autosave + --continue. + +Drives ``run_intake`` with the LLM call and terminal input stubbed, so we can +assert the REPL persists the conversation each turn and that ``--continue`` +reloads the newest unfinished one and keeps appending to the same record. +""" + +from __future__ import annotations + +import asyncio + +from arbor.cli.intake import repl +from arbor.cli.intake.conversation_store import find_conversations, load_messages + + +class _FakeProvider: + model = "fake-model" + base_url = None + + async def create(self, **_kw): # pragma: no cover - Agent.run is stubbed + raise AssertionError("provider.create should not be called; Agent.run is stubbed") + + def count_tokens(self, text: str) -> int: + return len(text.split()) + + +async def _fake_agent_run(self, user_message: str) -> str: + """Append a user + assistant turn, like the real loop would, minus the LLM.""" + self.messages.append({"role": "user", "content": user_message}) + self.messages.append({"role": "assistant", "content": [{"type": "text", "text": "ok"}]}) + return "ok" + + +def _scripted_reader(inputs): + queue = list(inputs) + + async def _reader(_session): + if not queue: + raise EOFError + return queue.pop(0) + + return _reader + + +class _NoopDisplay: + def __init__(self, *a, **k): + pass + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + +def _run(monkeypatch, cwd, inputs, *, continue_latest=False): + monkeypatch.setattr(repl.Agent, "run", _fake_agent_run) + monkeypatch.setattr(repl, "_read_user_line", _scripted_reader(inputs)) + monkeypatch.setattr(repl, "_build_session", lambda *a, **k: None) + monkeypatch.setattr(repl, "_print_welcome", lambda *a, **k: None) + monkeypatch.setattr(repl, "IntakeDisplay", _NoopDisplay) + return asyncio.run( + repl.run_intake( + provider=_FakeProvider(), + starting_cwd=cwd, + seed_message=None, + continue_latest=continue_latest, + ) + ) + + +def test_intake_autosaves_conversation_each_turn(tmp_path, monkeypatch): + outcome = _run(monkeypatch, tmp_path, ["hello arbor", "/quit"]) + assert outcome is None # /quit aborts without launching + + convs = find_conversations(tmp_path) + assert len(convs) == 1 + rec = convs[0] + assert rec.launched is False + assert rec.title.startswith("hello arbor") + user_msgs = [m.get("content") for m in load_messages(rec) if m.get("role") == "user"] + assert "hello arbor" in user_msgs + + +def test_continue_reloads_and_appends_to_same_conversation(tmp_path, monkeypatch): + _run(monkeypatch, tmp_path, ["first message", "/quit"]) + _run(monkeypatch, tmp_path, ["second message", "/quit"], continue_latest=True) + + convs = find_conversations(tmp_path) + assert len(convs) == 1 # continued the existing record, did not fork a new one + user_msgs = [m.get("content") for m in load_messages(convs[0]) if m.get("role") == "user"] + assert "first message" in user_msgs + assert "second message" in user_msgs + + +def test_continue_with_no_history_starts_fresh(tmp_path, monkeypatch): + # --continue with nothing to continue must not crash; it starts a fresh chat. + outcome = _run(monkeypatch, tmp_path, ["only message", "/quit"], continue_latest=True) + assert outcome is None + convs = find_conversations(tmp_path) + assert len(convs) == 1 + + +def test_visible_text_extracts_prose_and_skips_tool_plumbing(): + assert repl._visible_text("hi there") == "hi there" + assert repl._visible_text([{"type": "text", "text": "a"}, {"type": "tool_use", "name": "Bash"}]) == "a" + assert repl._visible_text([{"type": "tool_result", "content": "internal"}]) == "" + assert repl._visible_text(None) == "" + + +def _capture_history(monkeypatch, messages): + import io + + from rich.console import Console + + buf = io.StringIO() + monkeypatch.setattr(repl, "_console", Console(file=buf, width=200, soft_wrap=True)) + repl._print_resumed_history(messages) + return buf.getvalue() + + +def test_print_resumed_history_replays_user_and_assistant(monkeypatch): + out = _capture_history( + monkeypatch, + [ + {"role": "user", "content": "summarize the paper"}, + {"role": "assistant", "content": [{"type": "text", "text": "Here is the summary"}]}, + {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "x", "content": "internal"}]}, + {"role": "assistant", "content": [{"type": "tool_use", "name": "Bash", "input": {}}]}, + ], + ) + assert "summarize the paper" in out + assert "Here is the summary" in out + assert "previous conversation" in out + assert "internal" not in out # tool_result is plumbing, not shown + + +def test_print_resumed_history_empty_is_silent(monkeypatch): + assert _capture_history(monkeypatch, []) == "" From 990f660563a42674f72c88c6be834f07582f6f1a Mon Sep 17 00:00:00 2001 From: Xiaoyu <458788361@qq.com> Date: Thu, 25 Jun 2026 14:15:11 +0800 Subject: [PATCH 2/2] docs(intake): document conversation autosave, --continue, and /resume Add an "Outputs & Resume" section (EN + zh) covering the per-turn autosave under .arbor/conversations/, `arbor --continue` / `-C`, and the extended `/resume` that lists saved conversations alongside runs. Add an `arbor --continue` row to both README command tables. --- README.md | 1 + README.zh-CN.md | 1 + docs/outputs-and-resume.md | 24 ++++++++++++++++++++++++ docs/outputs-and-resume.zh.md | 22 ++++++++++++++++++++++ 4 files changed, 48 insertions(+) diff --git a/README.md b/README.md index 69e0ba7..4e11b96 100644 --- a/README.md +++ b/README.md @@ -402,6 +402,7 @@ Day to day you only need `arbor`: | Command | What it does | | --- | --- | | `arbor` | Start an interactive research session. | +| `arbor --continue` (`-C`) | Resume the most recent unfinished planning conversation in this directory (like `claude -c`). | | `arbor replay --demo` | Replay a bundled sample run in the live dashboard — no API key needed. Add `--html` for a shareable browser page. | | `arbor replay ` | Replay any past run's `events.jsonl` from its timeline (`--html` to export an interactive page). | | `arbor quickstart` | Get running fast with a free key (Gemini/Groq) or a local model (Ollama). | diff --git a/README.zh-CN.md b/README.zh-CN.md index 0796f96..3087b01 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -315,6 +315,7 @@ ROOT（基线：20%） | 命令 | 功能 | | ------------------------ | ----------------------------------------------------------- | | `arbor` | 启动交互式研究会话。 | +| `arbor --continue`（`-C`） | 续聊当前目录下最近一段未结束的规划对话（类似 `claude -c`）。 | | `arbor replay --demo` | 在实时仪表盘里回放一次内置示例运行 —— 无需 API 密钥；加 `--html` 生成可分享的网页。 | | `arbor replay ` | 按时间轴回放任意历史运行的 `events.jsonl`（`--html` 可导出交互式页面）。 | | `arbor quickstart` | 用免费密钥（Gemini/Groq）或本地模型（Ollama）快速跑起来。 | diff --git a/docs/outputs-and-resume.md b/docs/outputs-and-resume.md index 30b9dfe..0f76adf 100644 --- a/docs/outputs-and-resume.md +++ b/docs/outputs-and-resume.md @@ -41,6 +41,30 @@ Every experiment ran on its own git branch in an isolated worktree. Merged impro are on trunk; explored-but-unmerged ideas remain as branch refs you can inspect. During a run, `/branches` lists the explored branch refs and `/tree` prints the Idea Tree snapshot. +## Saving & resuming the planning conversation + +The intake chat — what you do at the `arbor` prompt *before* a run launches — is itself +saved every turn, so you can step away and pick it back up. Each conversation lives in its +own directory under the launch folder: + +```text +/.arbor/conversations// +``` + +with the message history (`messages.jsonl`) and a small `meta.json` (title, timestamps, +turn count). Saving is best-effort and never interrupts the chat. + +Two ways back in: + +- **`arbor --continue`** (`-C`) — reload the most recent *unfinished* conversation and keep + chatting, like `claude -c`. (`-c` is already `--config`, so the short flag is `-C`.) +- **`/resume`** inside intake — lists saved conversations (tagged `[chat]`) alongside + launched runs; pick a conversation to keep chatting, or a run to replay the engine. + +On resume the prior exchange is replayed to the terminal so you can see what was already +said. Once a conversation launches a run it is recorded as launched and dropped from +`--continue` — the run's own checkpoint (below) takes over. + ## Resuming an interrupted run If a run is interrupted — you stopped it, the machine rebooted, a budget tripped — resume diff --git a/docs/outputs-and-resume.zh.md b/docs/outputs-and-resume.zh.md index a338782..339b863 100644 --- a/docs/outputs-and-resume.zh.md +++ b/docs/outputs-and-resume.zh.md @@ -36,6 +36,28 @@ Arbor 记录一次运行产生的一切，让你能检视它、复现它，并 想法仍以分支 ref 的形式存在，供你检视。运行过程中，`/branches` 列出探索过的分支 ref，`/tree` 打印想法树快照。 +## 保存与续聊规划对话 + +`arbor` 启动后、真正发起运行*之前*的那段规划对话，本身也会每轮自动保存，方便你离开后接着聊。 +每段对话有自己的目录，位于启动目录下： + +```text +/.arbor/conversations// +``` + +里面是消息历史（`messages.jsonl`）和一个小的 `meta.json`（标题、时间戳、轮数）。保存是尽力而为， +绝不会打断对话。 + +两种回到对话的方式： + +- **`arbor --continue`**（`-C`）—— 重新载入最近一段*未结束*的对话并接着聊，类似 `claude -c`。 + （`-c` 已经是 `--config`，所以短旗用 `-C`。） +- 在 intake 里输入 **`/resume`** —— 会把已保存的对话（标 `[chat]`）和已启动的运行一起列出； + 选对话就继续聊，选运行就重放引擎。 + +续聊时会把之前的对话回放到终端，让你看到已经聊过的内容。一旦某段对话发起了运行，它会被标记为 +已启动并从 `--continue` 中排除——接力交给运行自己的检查点（见下文）。 + ## 续跑被中断的运行 如果一次运行被中断——你停了它、机器重启、预算耗尽——从它的检查点续跑，而不是从头再来：