CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## Unreleased
 
+- Added TUI session export/import command handling for JSON and Markdown.
+  `/session:export` now supports `--format md` and writes transcript-oriented
+  Markdown output, while `/session:import` restores sessions from JSON exports.
+  Updated TUI command/router Behave coverage and Robot smoke coverage for the
+  new session command paths. (#1004)
 - Expanded the TUI slash command overlay catalog to include 67 commands across
   14 groups, aligned with the specification command reference for session,
   persona, scope, plan, project, registry/config, context, and utility flows.

features/steps/tui_commands_coverage_steps.py

@@ -8,9 +8,12 @@ Targets uncovered lines in cleveragents.tui.commands:
 """
 
 import json
+import shutil
+import tempfile
 from dataclasses import dataclass, field
 from io import StringIO
 from pathlib import Path
+from typing import Any
 from unittest.mock import MagicMock, patch
 
 from behave import given, then, when
@@ -55,6 +58,65 @@ class FakePersonaState:
         return p.name if p else "default"
 
 
+@dataclass
+class FakeImportedSession:
+    """Minimal imported-session payload for command responses."""
+
+    session_id: str = "imported-session-001"
+    message_count: int = 2
+
+
+@dataclass
+class FakeSessionService:
+    """Session service stand-in supporting export/import."""
+
+    exported_session_id: str | None = None
+    imported_payload: dict[str, Any] | None = None
+
+    def export_session(self, session_id: str) -> dict[str, Any]:
+        self.exported_session_id = session_id
+        return {
+            "schema_version": "1.0",
+            "session_id": session_id,
+            "actor_name": "local/mock-default",
+            "namespace": "local",
+            "messages": [
+                {
+                    "message_id": "m-1",
+                    "role": "user",
+                    "content": "hello",
+                    "sequence": 0,
+                    "timestamp": "2026-01-01T00:00:00+00:00",
+                    "metadata": {},
+                    "tool_call_id": None,
+                },
+                {
+                    "message_id": "m-2",
+                    "role": "assistant",
+                    "content": "hi there",
+                    "sequence": 1,
+                    "timestamp": "2026-01-01T00:00:01+00:00",
+                    "metadata": {},
+                    "tool_call_id": None,
+                },
+            ],
+            "linked_plan_ids": [],
+            "token_usage": {
+                "input_tokens": 1,
+                "output_tokens": 2,
+                "estimated_cost": 0.0,
+            },
+            "metadata": {},
+            "created_at": "2026-01-01T00:00:00+00:00",
+            "updated_at": "2026-01-01T00:00:01+00:00",
+            "checksum": "dummy",
+        }
+
+    def import_session(self, data: dict[str, Any]) -> FakeImportedSession:
+        self.imported_payload = dict(data)
+        return FakeImportedSession()
+
+
 # ---------------------------------------------------------------------------
 # Background
 # ---------------------------------------------------------------------------
@@ -73,9 +135,11 @@ def step_router_with_mock_deps(context):
     """Create a router with an empty registry and basic state."""
     context.registry = FakePersonaRegistry()
     context.state = FakePersonaState()
+    context.session_service = FakeSessionService()
     context.router = TuiCommandRouter(
         persona_registry=context.registry,
         persona_state=context.state,
+        session_service=context.session_service,
     )
 
 
@@ -84,9 +148,11 @@ def step_router_with_set_support(context):
     """Create a router whose state can set an active persona."""
     context.registry = FakePersonaRegistry()
     context.state = FakePersonaState()
+    context.session_service = FakeSessionService()
     context.router = TuiCommandRouter(
         persona_registry=context.registry,
         persona_state=context.state,
+        session_service=context.session_service,
     )
 
 
@@ -97,9 +163,11 @@ def step_router_with_two_personas(context):
         _personas=[FakePersona(name="Alice"), FakePersona(name="Bob")]
     )
     context.state = FakePersonaState()
+    context.session_service = FakeSessionService()
     context.router = TuiCommandRouter(
         persona_registry=context.registry,
         persona_state=context.state,
+        session_service=context.session_service,
     )
 
 
@@ -108,9 +176,11 @@ def step_router_with_empty_registry(context):
     """Create a router whose registry returns no personas."""
     context.registry = FakePersonaRegistry(_personas=[])
     context.state = FakePersonaState()
+    context.session_service = FakeSessionService()
     context.router = TuiCommandRouter(
         persona_registry=context.registry,
         persona_state=context.state,
+        session_service=context.session_service,
     )
 
 
@@ -148,6 +218,66 @@ def step_handle_result_starts_with(context, prefix):
     )
 
 
+@then('the handle result should contain "{fragment}"')
+def step_handle_result_contains(context, fragment):
+    assert fragment in context.handle_result, (
+        f"Expected result to contain {fragment!r}, got {context.handle_result!r}"
+    )
+
+
+@when("I export the current session as markdown to a temporary file")
+def step_export_session_markdown_to_temp_file(context):
+    temp_dir = Path(tempfile.mkdtemp(prefix="tui-session-export-", dir=Path.cwd()))
+    context.add_cleanup(lambda: shutil.rmtree(temp_dir, ignore_errors=True))
+    relative_dir = temp_dir.relative_to(Path.cwd())
+    relative_path = relative_dir / "session.md"
+    context._session_markdown_path = Path.cwd() / relative_path
+    context.handle_result = context.router.handle(
+        f"session:export --format md {relative_path.as_posix()}",
+        session_id="test-session",
+    )
+
+
+@then("markdown session export should be written")
+def step_verify_markdown_export(context):
+    assert context._session_markdown_path.exists()
+    payload = context._session_markdown_path.read_text(encoding="utf-8")
+    assert "# Session Transcript" in payload
+    assert "## Transcript" in payload
+    assert "Session exported to" in context.handle_result
+
+
+@when("I import a session from a temporary JSON file")
+def step_import_session_from_temp_json(context):
+    temp_dir = Path(tempfile.mkdtemp(prefix="tui-session-import-", dir=Path.cwd()))
+    context.add_cleanup(lambda: shutil.rmtree(temp_dir, ignore_errors=True))
+    relative_dir = temp_dir.relative_to(Path.cwd())
+    relative_path = relative_dir / "session.json"
+    payload = {
+        "schema_version": "1.0",
+        "messages": [],
+        "checksum": "dummy",
+    }
+    (Path.cwd() / relative_path).write_text(json.dumps(payload), encoding="utf-8")
+    context.handle_result = context.router.handle(
+        f"session:import {relative_path.as_posix()}",
+        session_id="test-session",
+    )
+
+
+@when("I import a session from an invalid JSON file")
+def step_import_session_from_invalid_json(context):
+    temp_dir = Path(tempfile.mkdtemp(prefix="tui-session-import-bad-", dir=Path.cwd()))
+    context.add_cleanup(lambda: shutil.rmtree(temp_dir, ignore_errors=True))
+    relative_dir = temp_dir.relative_to(Path.cwd())
+    relative_path = relative_dir / "bad-session.json"
+    (Path.cwd() / relative_path).write_text("{invalid", encoding="utf-8")
+    context.handle_result = context.router.handle(
+        f"session:import {relative_path.as_posix()}",
+        session_id="test-session",
+    )
+
+
 # ---------------------------------------------------------------------------
 # run_tui() headless scenario
 # ---------------------------------------------------------------------------
@@ -168,6 +298,7 @@ def step_run_tui_headless(context):
     mock_container = MagicMock()
     mock_container.persona_registry.return_value = mock_registry
     mock_container.persona_state.return_value = mock_state
+    mock_container.session_service.return_value = MagicMock()
 
     captured = StringIO()
     with (
@@ -189,6 +320,7 @@ def step_run_tui_non_headless(context):
     mock_container = MagicMock()
     mock_container.persona_registry.return_value = mock_registry
     mock_container.persona_state.return_value = mock_state
+    mock_container.session_service.return_value = MagicMock()
 
     mock_app = MagicMock()
     mock_app.run = MagicMock()

features/tui_commands_coverage.feature

@@ -66,6 +66,31 @@ Feature: TUI Command Router and run_tui coverage
     When I call handle with raw input "session"
     Then the handle result should be "Current session: test-session"
 
+  Scenario: session colon command alias shows current session
+    Given a TuiCommandRouter with a mock registry and state
+    When I call handle with raw input "session:show"
+    Then the handle result should be "Current session: test-session"
+
+  Scenario: session export returns JSON payload by default
+    Given a TuiCommandRouter with a mock registry and state
+    When I call handle with raw input "session export"
+    Then the handle result should contain "schema_version"
+
+  Scenario: session export writes markdown transcript to file
+    Given a TuiCommandRouter with a mock registry and state
+    When I export the current session as markdown to a temporary file
+    Then markdown session export should be written
+
+  Scenario: session import reads JSON file and reports imported session
+    Given a TuiCommandRouter with a mock registry and state
+    When I import a session from a temporary JSON file
+    Then the handle result should start with "Session imported:"
+
+  Scenario: session import with invalid JSON returns an error
+    Given a TuiCommandRouter with a mock registry and state
+    When I import a session from an invalid JSON file
+    Then the handle result should start with "Session import failed: invalid JSON"
+
   Scenario: session unknown subcommand returns error
     Given a TuiCommandRouter with a mock registry and state
     When I call handle with raw input "session reset abc"

robot/tui_smoke.robot

@@ -40,3 +40,33 @@ TUI Headless Includes Router Help Payload
     Should Be Equal As Integers    ${result.rc}    0
     Should Contain    ${result.stdout}    "help"
     Should Contain    ${result.stdout}    /persona
+
+TUI Session Export Import Commands
+    ${script}=    Catenate    SEPARATOR=\n
+    ...    import json
+    ...    import shutil
+    ...    import tempfile
+    ...    from pathlib import Path
+    ...    from types import SimpleNamespace
+    ...    from cleveragents.tui.commands import TuiCommandRouter
+    ...    imported = SimpleNamespace(session_id="imported-session-robot", message_count=0)
+    ...    export_payload = {"schema_version": "1.0", "session_id": "sess-1", "actor_name": "local/mock-default", "namespace": "local", "messages": [{"role": "user", "content": "hello", "timestamp": "2026-01-01T00:00:00+00:00"}], "linked_plan_ids": [], "token_usage": {"input_tokens": 0, "output_tokens": 0, "estimated_cost": 0.0}, "metadata": {}, "created_at": "2026-01-01T00:00:00+00:00", "updated_at": "2026-01-01T00:00:00+00:00", "checksum": "dummy"}
+    ...    session_service = SimpleNamespace(export_session=lambda session_id: dict(export_payload, session_id=session_id), import_session=lambda data: imported)
+    ...    registry = SimpleNamespace(list_personas=lambda: [])
+    ...    state = SimpleNamespace(set_active_persona=lambda session_id, name: (_ for _ in ()).throw(ValueError("unused")))
+    ...    router = TuiCommandRouter(persona_registry=registry, persona_state=state, session_service=session_service)
+    ...    temp_dir = Path(tempfile.mkdtemp(prefix="tui-session-router-", dir=Path.cwd()))
+    ...    rel_md = temp_dir.relative_to(Path.cwd()) / "session.md"
+    ...    export_result = router.handle(f"session:export --format md {rel_md.as_posix()}", session_id="sess-1")
+    ...    assert "Session exported to" in export_result
+    ...    md_payload = (Path.cwd() / rel_md).read_text(encoding="utf-8")
+    ...    assert "# Session Transcript" in md_payload
+    ...    rel_json = temp_dir.relative_to(Path.cwd()) / "session.json"
+    ...    (Path.cwd() / rel_json).write_text(json.dumps({"schema_version": "1.0", "checksum": "dummy", "messages": []}), encoding="utf-8")
+    ...    import_result = router.handle(f"session:import {rel_json.as_posix()}", session_id="sess-1")
+    ...    assert "Session imported:" in import_result
+    ...    shutil.rmtree(temp_dir, ignore_errors=True)
+    ...    print("tui-session-router-ok")
+    ${result}=    Run Process    ${PYTHON}    -c    ${script}    shell=False    stderr=STDOUT
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tui-session-router-ok

src/cleveragents/tui/commands.py

@@ -4,8 +4,15 @@ from __future__ import annotations
 
 import json
 from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
 
 from cleveragents.application.container import get_container
+from cleveragents.domain.models.core.session import (
+    SessionImportError,
+    SessionNotFoundError,
+    SessionService,
+)
 from cleveragents.tui.app import CleverAgentsTuiApp, textual_available
 from cleveragents.tui.persona.registry import PersonaRegistry
 from cleveragents.tui.persona.state import PersonaState
@@ -17,19 +24,32 @@ class TuiCommandRouter:
 
     persona_registry: PersonaRegistry
     persona_state: PersonaState
+    session_service: SessionService | None = None
 
     def handle(self, raw: str, *, session_id: str) -> str:
         tokens = raw.strip().split()
         if not tokens:
             return "Empty command"
-        if tokens[0] == "persona":
-            return self._persona_command(tokens[1:], session_id=session_id)
-        if tokens[0] == "session":
-            return self._session_command(tokens[1:], session_id=session_id)
-        if tokens[0] == "help":
+        command, rest = self._split_command(tokens)
+        if command == "persona":
+            return self._persona_command(rest, session_id=session_id)
+        if command == "session":
+            return self._session_command(rest, session_id=session_id)
+        if command == "help":
             return "Commands: /persona, /session, /help"
         return f"Unknown command: /{raw}"
 
+    @staticmethod
+    def _split_command(tokens: list[str]) -> tuple[str, list[str]]:
+        command = tokens[0]
+        if ":" in command:
+            root, subcommand = command.split(":", maxsplit=1)
+            if root:
+                rest = [subcommand] if subcommand else []
+                rest.extend(tokens[1:])
+                return root, rest
+        return command, tokens[1:]
+
     def _persona_command(self, tokens: list[str], *, session_id: str) -> str:
         if not tokens or tokens[0] == "list":
             names = [persona.name for persona in self.persona_registry.list_personas()]
@@ -41,19 +61,204 @@ class TuiCommandRouter:
             return f"Active persona: {persona.name}"
         return f"Unknown persona command: {' '.join(tokens)}"
 
-    @staticmethod
-    def _session_command(tokens: list[str], *, session_id: str) -> str:
+    def _session_command(self, tokens: list[str], *, session_id: str) -> str:
         if not tokens or tokens[0] == "show":
             return f"Current session: {session_id}"
+        if tokens[0] == "export":
+            return self._session_export(tokens[1:], session_id=session_id)
+        if tokens[0] == "import":
+            return self._session_import(tokens[1:])
         return f"Unknown session command: {' '.join(tokens)}"
 
+    @staticmethod
+    def _resolve_relative_path(path: Path) -> Path:
+        if path.is_absolute():
+            raise ValueError("Path must be relative to current working directory")
+        base = Path.cwd().resolve()
+        resolved = (base / path).resolve()
+        if not resolved.is_relative_to(base):
+            raise ValueError("Path must stay within current working directory")
+        return resolved
+
+    @staticmethod
+    def _render_session_markdown(export_data: dict[str, Any]) -> str:
+        lines: list[str] = ["# Session Transcript", ""]
+        session_id = str(export_data.get("session_id", "(unknown)"))
+        actor = str(export_data.get("actor_name") or "(none)")
+        namespace = str(export_data.get("namespace") or "local")
+        created_at = str(export_data.get("created_at") or "")
+        updated_at = str(export_data.get("updated_at") or "")
+        lines.extend(
+            [
+                "## Metadata",
+                f"- Session ID: `{session_id}`",
+                f"- Actor: `{actor}`",
+                f"- Namespace: `{namespace}`",
+                f"- Created At: `{created_at}`",
+                f"- Updated At: `{updated_at}`",
+            ]
+        )
+
+        linked_plan_ids = export_data.get("linked_plan_ids", [])
+        if isinstance(linked_plan_ids, list) and linked_plan_ids:
+            plans = ", ".join(f"`{item!s}`" for item in linked_plan_ids)
+            lines.append(f"- Linked Plans: {plans}")
+
+        lines.extend(["", "## Transcript", ""])
+        messages = export_data.get("messages", [])
+        if not isinstance(messages, list) or not messages:
+            lines.append("(no messages)")
+        else:
+            for message in messages:
+                if not isinstance(message, dict):
+                    continue
+                role = str(message.get("role") or "unknown").upper()
+                timestamp = str(message.get("timestamp") or "")
+                content = str(message.get("content") or "").strip()
+                lines.append(f"### {role}")
+                if timestamp:
+                    lines.append(f"_{timestamp}_")
+                lines.append("")
+                if content:
+                    lines.extend(content.splitlines())
+                else:
+                    lines.append("(empty)")
+                lines.append("")
+
+        return "\n".join(lines).rstrip() + "\n"
+
+    @staticmethod
+    def _parse_export_args(args: list[str]) -> tuple[Path | None, str]:
+        output_path: Path | None = None
+        output_format = "json"
+        idx = 0
+        while idx < len(args):
+            arg = args[idx]
+            if arg == "--format":
+                idx += 1
+                if idx >= len(args):
+                    raise ValueError("--format requires a value")
+                output_format = args[idx].strip().lower()
+            elif arg.startswith("--format="):
+                output_format = arg.split("=", maxsplit=1)[1].strip().lower()
+            elif arg.startswith("--"):
+                raise ValueError(f"Unknown option: {arg}")
+            else:
+                if output_path is not None:
+                    raise ValueError("Only one output path is allowed")
+                output_path = Path(arg)
+            idx += 1
+
+        if output_format in {"md", "markdown"}:
+            return output_path, "md"
+        if output_format == "json":
+            return output_path, "json"
+        raise ValueError("--format must be one of: json, md")
+
+    @staticmethod
+    def _parse_import_args(args: list[str]) -> Path:
+        input_path: Path | None = None
+        idx = 0
+        while idx < len(args):
+            arg = args[idx]
+            if arg == "--input":
+                idx += 1
+                if idx >= len(args):
+                    raise ValueError("--input requires a file path")
+                input_path = Path(args[idx])
+            elif arg.startswith("--input="):
+                input_path = Path(arg.split("=", maxsplit=1)[1])
+            elif arg.startswith("--"):
+                raise ValueError(f"Unknown option: {arg}")
+            else:
+                if input_path is not None:
+                    raise ValueError("Only one input path is allowed")
+                input_path = Path(arg)
+            idx += 1
+
+        if input_path is None:
+            raise ValueError("Missing input path")
+        return input_path
+
+    def _session_export(self, args: list[str], *, session_id: str) -> str:
+        if self.session_service is None:
+            return "Session export unavailable: session service not configured"
+
+        try:
+            output_path, output_format = self._parse_export_args(args)
+        except ValueError as exc:
+            return f"Usage: /session:export [path] [--format json|md] ({exc})"
+
+        try:
+            export_data = self.session_service.export_session(session_id)
+        except SessionNotFoundError:
+            return f"Session not found: {session_id}"
+
+        rendered = (
+            json.dumps(export_data, indent=2, default=str)
+            if output_format == "json"
+            else self._render_session_markdown(export_data)
+        )
+
+        if output_path is None:
+            return rendered
+
+        try:
+            safe_output = self._resolve_relative_path(output_path)
+            safe_output.parent.mkdir(parents=True, exist_ok=True)
+            safe_output.write_text(rendered, encoding="utf-8")
+        except (OSError, ValueError) as exc:
+            return f"Session export failed: {exc}"
+
+        label = "JSON" if output_format == "json" else "Markdown"
+        return f"Session exported to {safe_output} ({label})"
+
+    def _session_import(self, args: list[str]) -> str:
+        if self.session_service is None:
+            return "Session import unavailable: session service not configured"
+
+        try:
+            input_path = self._parse_import_args(args)
+            safe_input = self._resolve_relative_path(input_path)
+        except ValueError as exc:
+            return f"Usage: /session:import <path> ({exc})"
+
+        if not safe_input.exists():
+            return f"Session import failed: file not found: {safe_input}"
+
+        try:
+            raw = safe_input.read_text(encoding="utf-8")
+            data = json.loads(raw)
+        except OSError as exc:
+            return f"Session import failed: {exc}"
+        except json.JSONDecodeError as exc:
+            return f"Session import failed: invalid JSON ({exc})"
+
+        if not isinstance(data, dict):
+            return "Session import failed: top-level JSON value must be an object"
+
+        try:
+            imported = self.session_service.import_session(data)
+        except SessionImportError as exc:
+            return f"Session import failed: {exc}"
+
+        return (
+            f"Session imported: {imported.session_id} "
+            f"({imported.message_count} messages)"
+        )
+
 
 def run_tui(*, headless: bool = False) -> int:
     """Run the Textual TUI app or a headless startup check."""
     container = get_container()
     registry = container.persona_registry()
     state = container.persona_state(registry=registry)
-    router = TuiCommandRouter(persona_registry=registry, persona_state=state)
+    session_service = container.session_service()
+    router = TuiCommandRouter(
+        persona_registry=registry,
+        persona_state=state,
+        session_service=session_service,
+    )
 
     if headless:
         payload = {

src/cleveragents/tui/slash_catalog.py

@@ -22,7 +22,7 @@ SLASH_COMMAND_SPECS: tuple[SlashCommandSpec, ...] = (
     SlashCommandSpec("session:close", "Session", "Close the current session"),
     SlashCommandSpec("session:delete", "Session", "Delete a saved session"),
     SlashCommandSpec("session:rename", "Session", "Rename current session"),
-    SlashCommandSpec("session:export", "Session", "Export session to JSON"),
+    SlashCommandSpec("session:export", "Session", "Export session (JSON or Markdown)"),
     SlashCommandSpec("session:import", "Session", "Import session from JSON"),
     SlashCommandSpec("persona:list", "Persona", "Display all personas"),
     SlashCommandSpec("persona:set", "Persona", "Switch active persona"),