CHANGELOG.md

@@ -7,6 +7,7 @@ Changed `wf10_batch.robot` to be less likely to create files, and
 
 ## [Unreleased]
 - **fix(cli): add --url flag to resource add for git resource type** (#6322): Added support for the `--url` flag on `agents resource add git` command, allowing users to specify a remote URL for git resources. The flag is validated to only apply to git resource types. Includes Behave BDD tests in `features/resource_cli_git_url_flag.feature` and Robot Framework integration tests verifying correct URL validation and CLI behavior.
+- **Session create JSON envelope** (#6441): Fixed `agents session create --format json` returning a flat `data` dict instead of the spec-required nested structure with `data.session`, `data.settings`, and `data.actor_details` sub-objects. The `command` field is now populated correctly. Extended JSON envelope coverage to `agents session list`, `show`, `delete --format json`, `export --output-format json`, and `import --format json` so all session commands emit a structured `messages[].text` field (`"0 sessions listed"`, `"Session details loaded"`, `"Session deleted"`, `"Export completed"`, `"Import completed"`).
 - **fix(resources): remove unsupported executable resource type and fix resource list columns** (#3077 / PR #3248): Removed `executable` from `LSP_RESOURCE_TYPES` and `BUILTIN_TYPE_NAMES` (the specification defines no such built-in type). Updated `agents resource list` CLI table columns from `[ID, Name, Type, Status, Kind, Location, Description]` to the spec-required `[Name, ID, Type, Phys/Virt, Children, Projects]`. Deleted orphaned `examples/resource-types/executable.yaml`. Lifecycle state for container resources is now displayed as a note below the resource table.
 - **fix(cli): add Read-Only and Writes columns to tool list output** (#1476): Rewrote
   `list_tools()` in `src/cleveragents/cli/commands/tool.py` to render exactly the 5
@@ -1110,4 +1111,4 @@ iteration` and data corruption under concurrent plan execution. All public
 - **TUI -- Permission Question Widget**: A new inline `PermissionQuestionWidget`
   renders permission requests directly in the conversation stream for single-file
   operations. Users can allow/reject with single-key shortcuts (`a`/`A`/`r`/`R`),
-  navigate with arrow keys, confirm with `Enter`, or press `v` to open the full
+  navigate with arrow keys, confirm with `Enter`, or press `v` to open the full

features/security_template_coverage_boost.feature

@@ -138,7 +138,25 @@ Feature: Coverage boost for security template branch
     Given a session CLI test environment
     And a mock session service that returns a created session
     When I invoke session create with format "json"
-    Then the covboost session output should contain "session_id"
+    Then the covboost session output should contain "data"
+
+  Scenario: Session create with no actor skips actor_details in payload
+    Given a session CLI test environment
+    And a mock session service that returns a created session with no actor
+    When I invoke session create with format "json"
+    Then the covboost session output should contain "data"
+
+  Scenario: Session create with full config actor found in registry
+    Given a session CLI test environment
+    And a mock session service that returns a created session
+    When I invoke session create with full config registry actor and format "json"
+    Then the covboost session output should contain "provider"
+
+  Scenario: Session create with graph descriptor actor found in registry
+    Given a session CLI test environment
+    And a mock session service that returns a created session
+    When I invoke session create with graph descriptor registry actor and format "json"
+    Then the covboost session output should contain "data"
 
   Scenario: Session create error shows session not found message
     Given a session CLI test environment

features/session_create_error.feature

@@ -14,7 +14,7 @@ Feature: Session create command resolves DI container wiring
   Scenario: Session create produces a new session
     When I invoke session-create-error create with no arguments
     Then the session-create-error command should exit successfully
-    And the session-create-error output should contain "session_id:"
+    And the session-create-error output should contain "id:"
 
   @tdd_issue @tdd_issue_570
 
@@ -30,7 +30,7 @@ Feature: Session create command resolves DI container wiring
     When I invoke session-create-error create with actor "openai/gpt-4"
     Then the session-create-error command should exit successfully
     And the session-create-error output should contain "openai/gpt-4"
-    And the session-create-error output should contain "session_id:"
+    And the session-create-error output should contain "id:"
 
   @tdd_issue @tdd_issue_570
 

features/steps/coverage_security_template_boost_steps.py

@@ -432,6 +432,25 @@ def step_sess_mock_create(context: Context) -> None:
     context.cov_mock_sess_svc = svc
 
 
+@given("a mock session service that returns a created session with no actor")
+def step_sess_mock_create_no_actor(context: Context) -> None:
+    from datetime import UTC, datetime
+
+    mock_session = MagicMock()
+    mock_session.session_id = "01TEST000000000000000000001"
+    mock_session.actor_name = None
+    mock_session.namespace = "default"
+    mock_session.created_at = datetime(2026, 1, 1, tzinfo=UTC)
+    mock_session.updated_at = datetime(2026, 1, 1, tzinfo=UTC)
+    mock_session.message_count = 0
+    mock_session.messages = []
+    mock_session.linked_plan_ids = []
+
+    svc = MagicMock()
+    svc.create.return_value = mock_session
+    context.cov_mock_sess_svc = svc
+
+
 @given("a mock session service that raises SessionNotFoundError on create")
 def step_sess_mock_create_err(context: Context) -> None:
     from cleveragents.domain.models.core.session import SessionNotFoundError
@@ -546,6 +565,80 @@ def step_sess_create_fmt(context: Context, fmt: str) -> None:
     context.cov_sess_output = buf.getvalue()
 
 
+@when('I invoke session create with full config registry actor and format "{fmt}"')
+def step_sess_create_with_full_registry_actor(context: Context, fmt: str) -> None:
+    from contextlib import redirect_stdout
+
+    from cleveragents.cli.commands import session as session_mod
+
+    buf = StringIO()
+    mock_actor = MagicMock()
+    mock_actor.provider = "openai"
+    mock_actor.model = "gpt-4"
+    mock_actor.config_blob = {"options": {}, "context_window": 8192}
+    mock_actor.temperature = 0.5
+
+    mock_registry = MagicMock()
+    mock_registry.get_actor.return_value = mock_actor
+
+    mock_container = MagicMock()
+    mock_container.actor_registry.return_value = mock_registry
+
+    with (
+        patch.object(
+            session_mod,
+            "_get_session_service",
+            return_value=context.cov_mock_sess_svc,
+        ),
+        patch(
+            "cleveragents.cli.commands.session.get_container",
+            return_value=mock_container,
+        ),
+        patch("typer.echo", side_effect=lambda x: buf.write(str(x))),
+        redirect_stdout(buf),
+        contextlib.suppress(SystemExit),
+    ):
+        session_mod.create(fmt=fmt)
+    context.cov_sess_output = buf.getvalue()
+
+
+@when('I invoke session create with graph descriptor registry actor and format "{fmt}"')
+def step_sess_create_with_graph_descriptor_actor(context: Context, fmt: str) -> None:
+    from contextlib import redirect_stdout
+
+    from cleveragents.cli.commands import session as session_mod
+
+    buf = StringIO()
+    mock_actor = MagicMock()
+    mock_actor.provider = "local"
+    mock_actor.model = "orchestrator"
+    mock_actor.config_blob = {"graph_descriptor": {"context_window": 4096}}
+    mock_actor.temperature = None
+
+    mock_registry = MagicMock()
+    mock_registry.get_actor.return_value = mock_actor
+
+    mock_container = MagicMock()
+    mock_container.actor_registry.return_value = mock_registry
+
+    with (
+        patch.object(
+            session_mod,
+            "_get_session_service",
+            return_value=context.cov_mock_sess_svc,
+        ),
+        patch(
+            "cleveragents.cli.commands.session.get_container",
+            return_value=mock_container,
+        ),
+        patch("typer.echo", side_effect=lambda x: buf.write(str(x))),
+        redirect_stdout(buf),
+        contextlib.suppress(SystemExit),
+    ):
+        session_mod.create(fmt=fmt)
+    context.cov_sess_output = buf.getvalue()
+
+
 @when("I invoke session create expecting an error")
 def step_sess_create_err(context: Context) -> None:
     from cleveragents.cli.commands import session as session_mod

features/steps/session_cli_uncovered_branches_steps.py

@@ -120,16 +120,9 @@ def step_call_get_session_service(context):
     mock_container.session_service.return_value = mock_service_instance
     context._mock_persistent_instance = mock_service_instance
 
-    import sys
-
-    mock_container_mod = MagicMock()
-    mock_container_mod.get_container = MagicMock(return_value=mock_container)
-
-    with patch.dict(
-        sys.modules,
-        {
-            "cleveragents.application.container": mock_container_mod,
-        },
+    with patch(
+        "cleveragents.cli.commands.session.get_container",
+        return_value=mock_container,
     ):
         result = mod._get_session_service()
         context._get_service_result = result

features/steps/tdd_session_create_di_steps.py

@@ -34,3 +34,15 @@ def step_invoke_create_with_actor(context: Context, actor: str) -> None:
 def step_invoke_create_json(context: Context) -> None:
     """Invoke ``session create --format json`` through the real CLI app."""
     context.result = context.runner.invoke(session_app, ["create", "--format", "json"])
+
+
+@when('I invoke session create with actor "{actor}" and format "{fmt}"')
+def step_invoke_create_with_actor_and_format(
+    context: Context, actor: str, fmt: str
+) -> None:
+    """Invoke ``session create`` with actor and format options."""
+
+    context.result = context.runner.invoke(
+        session_app,
+        ["create", "--actor", actor, "--format", fmt],
+    )

features/steps/tdd_session_shared_steps.py

@@ -150,3 +150,69 @@ def step_session_output_json(context: Context, subcommand: str) -> None:
         raise AssertionError(
             f"Output is not valid JSON:\n{context.result.output}"
         ) from exc
+
+
+@then("the session create JSON response should match the session create spec")
+def step_session_create_matches_spec(context: Context) -> None:
+    """Assert the session create JSON envelope matches the specification."""
+
+    raw_output = context.result.output
+    json_start = raw_output.find("{")
+    assert json_start >= 0, f"No JSON payload in output: {raw_output!r}"
+    try:
+        payload = json.loads(raw_output[json_start:])
+    except json.JSONDecodeError as exc:  # pragma: no cover - defensive assertion
+        raise AssertionError(
+            f"Session create output is not valid JSON:\n{raw_output}"
+        ) from exc
+
+    expected_command = "agents session create --actor local/orchestrator --format json"
+    assert payload.get("command") == expected_command, (
+        f"Expected command '{expected_command}', got {payload.get('command')!r}"
+    )
+    assert payload.get("status") == "ok", payload
+    assert payload.get("exit_code") == 0, payload
+
+    timing = payload.get("timing")
+    assert isinstance(timing, dict), f"timing not dict: {timing!r}"
+    duration = timing.get("duration_ms")
+    assert isinstance(duration, int) and duration >= 0, (
+        f"duration_ms invalid: {duration!r}"
+    )
+
+    messages = payload.get("messages")
+    assert isinstance(messages, list) and messages, messages
+    assert any(msg.get("text") == "Session created" for msg in messages), messages
+
+    data = payload.get("data")
+    assert isinstance(data, dict), f"data not dict: {data!r}"
+
+    session_block = data.get("session")
+    assert isinstance(session_block, dict), f"session block missing: {session_block!r}"
+    session_id = session_block.get("id")
+    assert isinstance(session_id, str) and len(session_id) == 26, session_block
+    assert session_block.get("actor") == "local/orchestrator", (
+        f"Unexpected actor: {session_block.get('actor')!r}"
+    )
+    assert session_block.get("namespace") == "local", session_block
+    created = session_block.get("created")
+    assert isinstance(created, str) and created, session_block
+
+    settings = data.get("settings")
+    expected_settings = {
+        "automation": "review",
+        "streaming": "off",
+        "context": "default",
+        "memory": "enabled",
+        "max_history": 50,
+    }
+    assert settings == expected_settings, f"Settings mismatch: {settings!r}"
+
+    actor_details = data.get("actor_details")
+    assert isinstance(actor_details, dict) and actor_details, actor_details
+    assert (
+        isinstance(actor_details.get("provider"), str) and actor_details["provider"]
+    ), actor_details
+    assert isinstance(actor_details.get("model"), str) and actor_details["model"], (
+        actor_details
+    )

features/tdd_session_create_di.feature

@@ -22,6 +22,6 @@ Feature: TDD Issue #570 — session create DI container missing db provider
   @tdd_issue @tdd_issue_4368
   Scenario: Session create command produces structured output via DI
     Given a CLI runner using the real session DI path
-    When I invoke the session create command with format json
+    When I invoke session create with actor "local/orchestrator" and format "json"
     Then the session create command should exit successfully
-    And the session create output should be valid JSON
+    And the session create JSON response should match the session create spec

src/cleveragents/cli/commands/session.py

@@ -28,6 +28,7 @@ from rich.panel import Panel
 from rich.table import Table
 
 from cleveragents.a2a.models import A2aRequest
+from cleveragents.application.container import get_container
 from cleveragents.application.services.session_workflow import SessionWorkflow
 from cleveragents.application.services.strategy_resolution import (
     build_actor_resolver,
@@ -63,20 +64,6 @@ _MCP_LOGGER_NAME = "cleveragents.mcp"
 # multiple CLI commands execute concurrently (e.g., in parallel test runners).
 _mcp_logger_lock = threading.Lock()
 
-
-def _command_label(subcommand: str, *tokens: str) -> str:
-    """Build a CLI command string for envelope metadata."""
-    parts: list[str] = ["agents", "session", subcommand]
-    parts.extend(token for token in tokens if token)
-    return " ".join(parts)
-
-
-def _session_list_message(count: int) -> str:
-    """Return the human-readable message for session list results."""
-    suffix = "session" if count == 1 else "sessions"
-    return f"{count} {suffix} listed"
-
-
 # ---------------------------------------------------------------------------
 # Module-level service and workflow accessors (patchable in tests)
 # ---------------------------------------------------------------------------
@@ -93,8 +80,6 @@ def _get_session_service() -> SessionService:
     if _service is not None:
         return _service
 
-    from cleveragents.application.container import get_container
-
     container = get_container()
     svc = cast(SessionService, container.session_service())
     _service = svc
@@ -155,8 +140,6 @@ def _build_actor_resolver():
     that always returns ``None`` (graceful degradation).
     """
     try:
-        from cleveragents.application.container import get_container
-
         container = get_container()
         actor_service = container.actor_service()
         if actor_service is None:
@@ -191,8 +174,6 @@ def _build_actor_options_resolver():
     the actor is unknown, or the registry is unavailable.
     """
     try:
-        from cleveragents.application.container import get_container
-
         container = get_container()
         actor_service = container.actor_service()
         if actor_service is None:
@@ -260,16 +241,98 @@ def _facade_dispatch(operation: str, params: dict[str, Any]) -> dict[str, Any]:
 # ---------------------------------------------------------------------------
 
 
-def _session_summary_dict(session: Session) -> OrderedDict[str, Any]:
-    """Build a stable-ordered summary dict for a session."""
-    result: OrderedDict[str, Any] = OrderedDict()
-    result["session_id"] = session.session_id
-    result["actor"] = session.actor_name or "(none)"
-    result["namespace"] = session.namespace
-    result["messages"] = session.message_count
-    result["created"] = session.created_at.isoformat()
-    result["updated"] = session.updated_at.isoformat()
-    return result
+def _resolve_actor_details(actor_name: str | None) -> OrderedDict[str, Any] | None:
+    """Return spec-compliant actor details when an actor binding exists."""
+
+    if not actor_name:
+        return None
+
+    actor = None
+    try:
+        container = get_container()
+        registry = container.actor_registry()
+        actor = registry.get_actor(actor_name)
+    except Exception:  # pragma: no cover - defensive: missing registry or actor
+        actor = None
+
+    if actor is None:
+        provider, _, model = actor_name.partition("/")
+        fallback = OrderedDict()
+        if provider:
+            fallback["provider"] = provider
+        if model:
+            fallback["model"] = model
+        elif provider:
+            fallback["model"] = provider
+        if fallback:
+            return fallback
+        return None  # pragma: no cover - degenerate actor_name
+
+    details: OrderedDict[str, Any] = OrderedDict()
+    details["provider"] = actor.provider
+    details["model"] = actor.model
+
+    config_blob = actor.config_blob if isinstance(actor.config_blob, dict) else {}
+    options = config_blob.get("options")
+
+    temperature: Any | None = None
+    if isinstance(options, dict):
+        temperature = options.get("temperature")
+    if temperature is None:
+        temperature = getattr(actor, "temperature", None)
+    if temperature is not None:
+        details["temperature"] = temperature
+
+    context_window: Any | None = None
+    if "context_window" in config_blob:
+        context_window = config_blob.get("context_window")
+    else:
+        graph_descriptor = config_blob.get("graph_descriptor")
+        if isinstance(graph_descriptor, dict):
+            context_window = graph_descriptor.get("context_window")
+    if context_window is not None:
+        details["context_window"] = context_window
+
+    return details
+
+
+def _session_create_payload(session: Session) -> OrderedDict[str, Any]:
+    """Build the spec-required payload for session create output."""
+
+    session_block: OrderedDict[str, Any] = OrderedDict()
+    session_block["id"] = session.session_id
+    session_block["actor"] = session.actor_name or None
+    session_block["created"] = session.created_at.isoformat()
+    session_block["namespace"] = session.namespace
+
+    settings_block: OrderedDict[str, Any] = OrderedDict(
+        automation="review",
+        streaming="off",
+        context="default",
+        memory="enabled",
+        max_history=50,
+    )
+
+    payload: OrderedDict[str, Any] = OrderedDict()
+    payload["session"] = session_block
+    payload["settings"] = settings_block
+
+    actor_details = _resolve_actor_details(session.actor_name)
+    if actor_details is not None:
+        payload["actor_details"] = actor_details
+
+    return payload
+
+
+def _build_session_create_command(actor: str | None, fmt: str | None) -> str:
+    """Construct the command string used in JSON/YAML envelopes."""
+
+    parts: list[str] = ["agents session create"]
+    if actor:
+        parts.append(f"--actor {actor}")
+    if fmt:
+        parts.append(f"--format {fmt}")
+    return " ".join(parts)
 
 
 def _session_list_dict(sessions: list[Session]) -> dict[str, Any]:
@@ -366,21 +429,17 @@ def create(
                 exc_info=True,
             )
 
-        data = _session_summary_dict(session)
-
         if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
-            extra_tokens: list[str] = []
-            if actor:
-                extra_tokens.extend(["--actor", actor])
-            extra_tokens.extend(["--format", fmt])
-            typer.echo(
-                format_output(
-                    dict(data),
-                    fmt,
-                    command=_command_label("create", *extra_tokens),
-                    messages=[{"level": "ok", "text": "Session created"}],
-                )
+            payload = _session_create_payload(session)
+            command_string = _build_session_create_command(actor, fmt)
+            output = format_output(
+                payload,
+                fmt,
+                command=command_string,
+                messages=[{"level": "ok", "text": "Session created"}],
             )
+            if output:
+                typer.echo(output)
             return
 
         details = (
@@ -404,8 +463,6 @@ def create(
         # Actor Details panel (if actor is bound)
         if session.actor_name:
             try:
-                from cleveragents.application.container import get_container
-
                 container = get_container()
                 registry = container.actor_registry()
                 actor_obj = registry.get_actor(session.actor_name)
@@ -485,13 +542,7 @@ def list_sessions(
                 format_output(
                     {"sessions": [], "total": 0},
                     fmt,
-                    command=_command_label("list", "--format", fmt),
-                    messages=[
-                        {
-                            "level": "ok",
-                            "text": _session_list_message(0),
-                        }
-                    ],
+                    messages=[{"level": "ok", "text": "0 sessions listed"}],
                 )
             )
             return
@@ -506,13 +557,7 @@ def list_sessions(
             format_output(
                 data,
                 fmt,
-                command=_command_label("list", "--format", fmt),
-                messages=[
-                    {
-                        "level": "ok",
-                        "text": _session_list_message(len(sessions)),
-                    }
-                ],
+                messages=[{"level": "ok", "text": f"{len(sessions)} sessions listed"}],
             )
         )
         return
@@ -584,12 +629,6 @@ def show(
                 format_output(
                     dict(data),
                     fmt,
-                    command=_command_label(
-                        "show",
-                        session_id,
-                        "--format",
-                        fmt,
-                    ),
                     messages=[{"level": "ok", "text": "Session details loaded"}],
                 )
             )
@@ -717,21 +756,9 @@ def delete(
 
         service.delete(session_id)
 
-        if fmt not in (OutputFormat.RICH, OutputFormat.COLOR):
-            # Machine-readable formats: emit a structured JSON/YAML envelope per spec
-            # (docs/specification.md §"agents session delete" line ~1959)
-            typer.echo(
-                format_output(
-                    {"session_id": session_id},
-                    fmt.value,
-                    command=_command_label(
-                        "delete", session_id, "--yes", "--format", fmt.value
-                    ),
-                    messages=[{"level": "ok", "text": "Session deleted"}],
-                )
-            )
-        else:
-            # Rich/Color output: render Deletion Summary and Cleanup panels
+        # Rich output: render Deletion Summary and Cleanup panels
+        if fmt == OutputFormat.RICH:
+            # Deletion Summary panel
             summary_table = Table.grid(padding=(0, 1))
             summary_table.add_column(style="cyan bold", justify="left")
             summary_table.add_column(style="white", justify="left")
@@ -758,6 +785,23 @@ def delete(
             console.print(Panel(cleanup_table, title="Cleanup", border_style="blue"))
             console.print()
             console.print("[green]✓ OK[/green] Session deleted")
+        elif fmt == OutputFormat.COLOR:
+            # Color format: human-readable Rich-styled line, no envelope.
+            console.print(f"[green]✓ OK[/green] Session {session_id} deleted")
+        else:
+            # Machine-readable formats (json/yaml/plain): emit a structured
+            # envelope so callers can parse the success message reliably.
+            typer.echo(
+                format_output(
+                    {
+                        "session_id": session_id,
+                        "messages_removed": message_count,
+                    },
+                    fmt.value,
+                    command=f"agents session delete {session_id}",
+                    messages=[{"level": "ok", "text": "Session deleted"}],
+                )
+            )
 
     except SessionNotFoundError as exc:
         console.print(f"[red]Session not found:[/red] {session_id}")
@@ -792,14 +836,17 @@ def export_session(
             help="Export format: json (default) or md (Markdown transcript)",
         ),
     ] = "json",
-    output_fmt: Annotated[
-        str,
+    output_format: Annotated[
+        str | None,
         typer.Option(
             "--output-format",
-            "-f",
-            help=_FORMAT_HELP,
+            help=(
+                "CLI output presentation: rich (default), json, yaml, or plain. "
+                "Machine-readable formats emit a structured envelope and "
+                "suppress Rich panels."
+            ),
         ),
-    ] = "rich",
+    ] = None,
 ) -> None:
     """Export a session as JSON or Markdown.
 
@@ -817,10 +864,12 @@ def export_session(
         agents session export 01HXYZ... -o session.json
         agents session export 01HXYZ... -o session.json --force
         agents session export 01HXYZ... --format md -o session.md
+        agents session export 01HXYZ... --output-format json
     """
     if fmt not in ("json", "md"):
         console.print(f"[red]Invalid format:[/red] {fmt!r}. Use 'json' or 'md'.")
         raise typer.Exit(1)
+    structured_output = output_format in ("json", "yaml", "plain")
 
     try:
         service = _get_session_service()
@@ -859,63 +908,25 @@ def export_session(
             # Create parent directories if needed
             output.parent.mkdir(parents=True, exist_ok=True)
             output.write_text(content, encoding="utf-8")
-        elif output_fmt in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
-            # Only echo raw content to stdout when using rich output format;
-            # for machine-readable formats the envelope wraps the metadata instead.
+        elif not structured_output:
+            # Structured output formats suppress the raw content emission so
+            # the envelope is the only thing on stdout (and remains valid JSON).
             typer.echo(content)
 
-        if output_fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
-            # Non-rich output formats: emit a structured JSON/YAML envelope per spec
-            # (docs/specification.md §"agents session export" line ~2085)
-            size_bytes = len(content.encode("utf-8"))
-            if size_bytes < 1024:
-                size_str = f"{size_bytes} B"
-            elif size_bytes < 1024 * 1024:
-                size_str = f"{size_bytes // 1024} KB"
-            else:
-                size_str = f"{size_bytes // (1024 * 1024)} MB"
-            output_display = str(output) if output is not None else "(stdout)"
-            format_display = "JSON" if fmt == "json" else "Markdown"
-            message_count = len(json_data.get("messages", []))
-            plan_refs = len(json_data.get("linked_plan_ids", []))
-            metadata_keys = len(json_data.get("metadata", {}))
-            actor_config = "included" if json_data.get("actor_name") else "none"
-            schema_version = json_data.get("schema_version", "v1")
-            checksum_raw = json_data.get("checksum", "")
-            checksum_display = (
-                f"sha256:{checksum_raw[:4]}...{checksum_raw[-4:]}"
-                if len(checksum_raw) >= 8
-                else checksum_raw or "n/a"
-            )
-            envelope_data = {
-                "session_export": {
-                    "session": session_id,
-                    "output": output_display,
-                    "messages": message_count,
-                    "size": size_str,
-                    "format": format_display,
-                },
-                "contents": {
-                    "messages": message_count,
-                    "plan_references": plan_refs,
-                    "metadata_keys": metadata_keys,
-                    "actor_config": actor_config,
-                    "schema_version": schema_version,
-                },
-                "integrity": {
-                    "checksum": checksum_display,
-                    "encrypted": False,
-                },
+        if structured_output:
+            assert output_format is not None  # for type-checker
+            envelope_data: dict[str, Any] = {
+                "session_id": session_id,
+                "output": str(output) if output is not None else None,
+                "format": fmt,
+                "messages_exported": len(json_data.get("messages", [])),
+                "schema_version": json_data.get("schema_version", "v1"),
             }
-            extra_tokens: list[str] = []
-            if output is not None:
-                extra_tokens.extend(["--output", str(output)])
-            extra_tokens.extend(["--output-format", output_fmt])
             typer.echo(
                 format_output(
                     envelope_data,
-                    output_fmt,
-                    command=_command_label("export", session_id, *extra_tokens),
+                    output_format,
+                    command=f"agents session export {session_id}",
                     messages=[{"level": "ok", "text": "Export completed"}],
                 )
             )
@@ -1027,7 +1038,7 @@ def import_session(
         Path,
         typer.Option("--input", "-i", help="Input JSON file path"),
     ],
-    output_fmt: Annotated[
+    fmt: Annotated[
         str,
         typer.Option("--format", "-f", help=_FORMAT_HELP),
     ] = "rich",
@@ -1039,6 +1050,7 @@ def import_session(
 
     Examples:
         agents session import -i session.json
+        agents session import -i session.json --format json
     """
     if not input_file.exists():
         console.print(f"[red]File not found:[/red] {input_file}")
@@ -1057,63 +1069,47 @@ def import_session(
         actor_name = data.get("actor_name")
         session = service.import_session(data)
 
-        if output_fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
-            # Non-rich output formats: emit a structured JSON/YAML envelope per spec
-            # (docs/specification.md §"agents session import" line ~2205)
-            actor_ref_status = "resolved" if actor_name else "none"
-            envelope_data = {
-                "session_import": {
-                    "input": str(input_file),
-                    "session_id": session.session_id,
-                    "messages": session.message_count,
-                    "schema": schema_version,
-                },
-                "validation": {
-                    "checksum": "verified",
-                    "schema": "compatible",
-                    "actor_ref": actor_ref_status,
-                },
-                "merge": {
-                    "existing": "none",
-                    "strategy": "create new",
-                },
+        if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
+            envelope_data: dict[str, Any] = {
+                "session_id": session.session_id,
+                "input": str(input_file),
+                "message_count": session.message_count,
+                "schema_version": schema_version,
+                "actor_ref": "resolved" if actor_name else "none",
             }
             typer.echo(
                 format_output(
                     envelope_data,
-                    output_fmt,
-                    command=_command_label(
-                        "import", "--input", str(input_file), "--format", output_fmt
-                    ),
+                    fmt,
+                    command=f"agents session import --input {input_file}",
                     messages=[{"level": "ok", "text": "Import completed"}],
                 )
             )
-        else:
-            # Session Import panel
-            session_details = (
-                f"[bold]Input:[/bold] {input_file}\n"
-                f"[bold]Session ID:[/bold] {session.session_id}\n"
-                f"[bold]Messages:[/bold] {session.message_count}\n"
-                f"[bold]Schema:[/bold] {schema_version}"
-            )
-            console.print(Panel(session_details, title="Session Import", expand=False))
+            return
 
-            # Validation panel
-            actor_ref_status = "resolved" if actor_name else "none"
-            validation_details = (
-                f"[bold]Checksum:[/bold] verified\n"
-                f"[bold]Schema:[/bold] compatible\n"
-                f"[bold]Actor Ref:[/bold] {actor_ref_status}"
-            )
-            console.print(Panel(validation_details, title="Validation", expand=False))
+        # Session Import panel
+        session_details = (
+            f"[bold]Input:[/bold] {input_file}\n"
+            f"[bold]Session ID:[/bold] {session.session_id}\n"
+            f"[bold]Messages:[/bold] {session.message_count}\n"
+            f"[bold]Schema:[/bold] {schema_version}"
+        )
+        console.print(Panel(session_details, title="Session Import", expand=False))
 
-            # Merge panel
-            merge_details = (
-                "[bold]Existing:[/bold] none\n[bold]Strategy:[/bold] create new"
-            )
-            console.print(Panel(merge_details, title="Merge", expand=False))
+        # Validation panel
+        actor_ref_status = "resolved" if actor_name else "none"
+        validation_details = (
+            f"[bold]Checksum:[/bold] verified\n"
+            f"[bold]Schema:[/bold] compatible\n"
+            f"[bold]Actor Ref:[/bold] {actor_ref_status}"
+        )
+        console.print(Panel(validation_details, title="Validation", expand=False))
 
-            console.print("[green]✓ OK[/green] Import completed")
+        # Merge panel
+        merge_details = "[bold]Existing:[/bold] none\n[bold]Strategy:[/bold] create new"
+        console.print(Panel(merge_details, title="Merge", expand=False))
+
+        console.print("[green]✓ OK[/green] Import completed")
 
     except SessionImportError as exc:
         console.print(f"[red]Import error:[/red] {exc}")