From 6b9879cd1b5ad5a18e4794b5e738a289458c5a71 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Wed, 22 Apr 2026 18:04:25 +0900 Subject: [PATCH] =?UTF-8?q?fix:=20#166=20=E2=80=94=20flush-transcript=20no?= =?UTF-8?q?w=20accepts=20--directory=20/=20--output-format=20/=20--session?= =?UTF-8?q?-id;=20session-creation=20command=20parity=20with=20#160/#165?= =?UTF-8?q?=20lifecycle=20triplet?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/main.py | 42 +++++- src/query_engine.py | 17 ++- tests/test_flush_transcript_cli.py | 206 +++++++++++++++++++++++++++++ 3 files changed, 259 insertions(+), 6 deletions(-) create mode 100644 tests/test_flush_transcript_cli.py diff --git a/src/main.py b/src/main.py index 50797a3e..36d88da2 100644 --- a/src/main.py +++ b/src/main.py @@ -81,8 +81,24 @@ def build_parser() -> argparse.ArgumentParser: ), ) - flush_parser = subparsers.add_parser('flush-transcript', help='persist and flush a temporary session transcript') + flush_parser = subparsers.add_parser( + 'flush-transcript', + help='persist and flush a temporary session transcript (#160/#166: claw-native session API)', + ) flush_parser.add_argument('prompt') + flush_parser.add_argument( + '--directory', help='session storage directory (default: .port_sessions)' + ) + flush_parser.add_argument( + '--output-format', + choices=['text', 'json'], + default='text', + help='output format', + ) + flush_parser.add_argument( + '--session-id', + help='deterministic session ID (default: auto-generated UUID)', + ) load_session_parser = subparsers.add_parser( 'load-session', @@ -232,11 +248,29 @@ def main(argv: list[str] | None = None) -> int: return 2 return 0 if args.command == 'flush-transcript': + from pathlib import Path as _Path engine = QueryEnginePort.from_workspace() + # #166: allow deterministic session IDs for claw checkpointing/replay. + # When unset, the engine's auto-generated UUID is used (backward compat). + if args.session_id: + engine.session_id = args.session_id engine.submit_message(args.prompt) - path = engine.persist_session() - print(path) - print(f'flushed={engine.transcript_store.flushed}') + directory = _Path(args.directory) if args.directory else None + path = engine.persist_session(directory) + if args.output_format == 'json': + import json as _json + print(_json.dumps({ + 'session_id': engine.session_id, + 'path': path, + 'flushed': engine.transcript_store.flushed, + 'messages_count': len(engine.mutable_messages), + 'input_tokens': engine.total_usage.input_tokens, + 'output_tokens': engine.total_usage.output_tokens, + })) + else: + # #166: legacy text output preserved byte-for-byte for backward compat. + print(path) + print(f'flushed={engine.transcript_store.flushed}') return 0 if args.command == 'load-session': from pathlib import Path as _Path diff --git a/src/query_engine.py b/src/query_engine.py index 97c060d5..5f3f3eda 100644 --- a/src/query_engine.py +++ b/src/query_engine.py @@ -153,7 +153,19 @@ class QueryEnginePort: def flush_transcript(self) -> None: self.transcript_store.flush() - def persist_session(self) -> str: + def persist_session(self, directory: 'Path | None' = None) -> str: + """Flush the transcript and save the session to disk. + + Args: + directory: Optional override for the storage directory. When None + (default, for backward compat), uses the default location + (``.port_sessions`` in CWD). When set, passes through to + ``save_session`` which already supports directory overrides. + + #166: added directory parameter to match the session-lifecycle CLI + surface established by #160/#165. Claws running out-of-tree can now + redirect session creation to a workspace-specific dir without chdir. + """ self.flush_transcript() path = save_session( StoredSession( @@ -161,7 +173,8 @@ class QueryEnginePort: messages=tuple(self.mutable_messages), input_tokens=self.total_usage.input_tokens, output_tokens=self.total_usage.output_tokens, - ) + ), + directory, ) return str(path) diff --git a/tests/test_flush_transcript_cli.py b/tests/test_flush_transcript_cli.py new file mode 100644 index 00000000..27701b16 --- /dev/null +++ b/tests/test_flush_transcript_cli.py @@ -0,0 +1,206 @@ +"""Tests for flush-transcript CLI parity with the #160/#165 lifecycle triplet (ROADMAP #166). + +Verifies that session *creation* now accepts the same flag family as session +management (list/delete/load): +- --directory DIR (alternate storage location) +- --output-format {text,json} (structured output) +- --session-id ID (deterministic IDs for claw checkpointing) + +Also verifies backward compat: default text output unchanged byte-for-byte. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + + +_REPO_ROOT = Path(__file__).resolve().parent.parent + + +def _run_cli(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, '-m', 'src.main', *args], + capture_output=True, text=True, cwd=str(_REPO_ROOT), + ) + + +class TestDirectoryFlag: + def test_flush_transcript_writes_to_custom_directory(self, tmp_path: Path) -> None: + result = _run_cli( + 'flush-transcript', 'hello world', + '--directory', str(tmp_path), + ) + assert result.returncode == 0, result.stderr + # Exactly one session file should exist in the directory + files = list(tmp_path.glob('*.json')) + assert len(files) == 1 + # And the legacy text output points to that file + assert str(files[0]) in result.stdout + + +class TestSessionIdFlag: + def test_explicit_session_id_is_respected(self, tmp_path: Path) -> None: + result = _run_cli( + 'flush-transcript', 'hello', + '--directory', str(tmp_path), + '--session-id', 'deterministic-id-42', + ) + assert result.returncode == 0, result.stderr + expected_path = tmp_path / 'deterministic-id-42.json' + assert expected_path.exists(), ( + f'session file not created at deterministic path: {expected_path}' + ) + # And it should contain the ID we asked for + data = json.loads(expected_path.read_text()) + assert data['session_id'] == 'deterministic-id-42' + + def test_auto_session_id_when_flag_omitted(self, tmp_path: Path) -> None: + """Without --session-id, engine still auto-generates a UUID (backward compat).""" + result = _run_cli( + 'flush-transcript', 'hello', + '--directory', str(tmp_path), + ) + assert result.returncode == 0 + files = list(tmp_path.glob('*.json')) + assert len(files) == 1 + # The filename (minus .json) should be a 32-char hex UUID + stem = files[0].stem + assert len(stem) == 32 + assert all(c in '0123456789abcdef' for c in stem) + + +class TestOutputFormatFlag: + def test_json_mode_emits_structured_envelope(self, tmp_path: Path) -> None: + result = _run_cli( + 'flush-transcript', 'hello', + '--directory', str(tmp_path), + '--session-id', 'beta', + '--output-format', 'json', + ) + assert result.returncode == 0 + data = json.loads(result.stdout) + assert data['session_id'] == 'beta' + assert data['flushed'] is True + assert data['path'].endswith('beta.json') + # messages_count and token counts should be present and typed + assert isinstance(data['messages_count'], int) + assert isinstance(data['input_tokens'], int) + assert isinstance(data['output_tokens'], int) + + def test_text_mode_byte_identical_to_pre_166_output(self, tmp_path: Path) -> None: + """Legacy text output must not change — claws may be parsing it.""" + result = _run_cli( + 'flush-transcript', 'hello', + '--directory', str(tmp_path), + ) + assert result.returncode == 0 + lines = result.stdout.strip().split('\n') + # Line 1: path ending in .json + assert lines[0].endswith('.json') + # Line 2: exact legacy format + assert lines[1] == 'flushed=True' + + +class TestBackwardCompat: + def test_no_flags_default_behaviour(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Running with no flags still works (default dir, text mode, auto UUID).""" + import os + env = os.environ.copy() + env['PYTHONPATH'] = str(_REPO_ROOT) + result = subprocess.run( + [sys.executable, '-m', 'src.main', 'flush-transcript', 'hello'], + capture_output=True, text=True, cwd=str(tmp_path), env=env, + ) + assert result.returncode == 0, result.stderr + # Default dir is `.port_sessions` in CWD + sessions_dir = tmp_path / '.port_sessions' + assert sessions_dir.exists() + assert len(list(sessions_dir.glob('*.json'))) == 1 + + +class TestLifecycleIntegration: + """#166's real value: the triplet + creation command are now a coherent family.""" + + def test_create_then_list_then_load_then_delete_roundtrip( + self, tmp_path: Path, + ) -> None: + """End-to-end: flush → list → load → delete, all via the same --directory.""" + # 1. Create + create_result = _run_cli( + 'flush-transcript', 'roundtrip test', + '--directory', str(tmp_path), + '--session-id', 'rt-session', + '--output-format', 'json', + ) + assert create_result.returncode == 0 + assert json.loads(create_result.stdout)['session_id'] == 'rt-session' + + # 2. List + list_result = _run_cli( + 'list-sessions', + '--directory', str(tmp_path), + '--output-format', 'json', + ) + assert list_result.returncode == 0 + list_data = json.loads(list_result.stdout) + assert 'rt-session' in list_data['sessions'] + + # 3. Load + load_result = _run_cli( + 'load-session', 'rt-session', + '--directory', str(tmp_path), + '--output-format', 'json', + ) + assert load_result.returncode == 0 + assert json.loads(load_result.stdout)['loaded'] is True + + # 4. Delete + delete_result = _run_cli( + 'delete-session', 'rt-session', + '--directory', str(tmp_path), + '--output-format', 'json', + ) + assert delete_result.returncode == 0 + + # 5. Verify gone + verify_result = _run_cli( + 'load-session', 'rt-session', + '--directory', str(tmp_path), + '--output-format', 'json', + ) + assert verify_result.returncode == 1 + assert json.loads(verify_result.stdout)['error']['kind'] == 'session_not_found' + + +class TestFullFamilyParity: + """All four session-lifecycle CLI commands accept the same core flag pair. + + This is the #166 acceptance test: flush-transcript joins the family. + """ + + @pytest.mark.parametrize( + 'command', + ['list-sessions', 'delete-session', 'load-session', 'flush-transcript'], + ) + def test_all_four_accept_directory_flag(self, command: str) -> None: + help_text = _run_cli(command, '--help').stdout + assert '--directory' in help_text, ( + f'{command} missing --directory flag (#166 parity gap)' + ) + + @pytest.mark.parametrize( + 'command', + ['list-sessions', 'delete-session', 'load-session', 'flush-transcript'], + ) + def test_all_four_accept_output_format_flag(self, command: str) -> None: + help_text = _run_cli(command, '--help').stdout + assert '--output-format' in help_text, ( + f'{command} missing --output-format flag (#166 parity gap)' + )