From a2834349b0ea0362f0922745c25d57b7fa975847 Mon Sep 17 00:00:00 2001
From: Patrick Buckley <eous@users.noreply.github.com>
Date: Fri, 29 May 2026 21:03:29 -0700
Subject: [PATCH] feat(export): export workstream conversations as OpenAI
 messages JSON
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add a workstream conversation export on three surfaces, all sharing one
serializer (turnstone/core/export.py):

- `turnstone-admin export <ws_id> [--children] [-o FILE|-]` — offline,
  direct-DB. `--children` bundles a coordinator's parent conversation
  plus one JSON per child into a zip (parent.json + children/<id>.json,
  no manifest).
- `GET /v1/api/workstreams/{ws_id}/export` — conversation-only file
  download, mounted on both the node (interactive) and console
  (coordinator) lifespans via `make_export_handler(cfg)`, reusing the
  /history gate ladder (permission_gate, tenant_check, list_kind
  cross-kind isolation) so ownership and isolation come for free.
- Web UI — an "Export conversation" item in the interactive per-tab
  dropdown (scoped to that tab's workstream) and an Export button on the
  coordinator appbar.

Format is OpenAI Chat Completions messages JSON (`{"messages": [...]}`),
built from `sanitize_messages(load_messages(repair=True))`. Persisted
reasoning is surfaced on assistant messages as a flat `reasoning_content`
field (the convention OpenAI-compatible inference servers use) via a
dedicated helper that runs before sanitize strips the internal
_provider_content lane. Attachments ride along as the standard image_url
/ inlined-document content parts.

Lets users get conversations out in a portable interchange format
(backup, fine-tuning datasets, sharing, interop) without lock-in.
Closes #613.

Non-obvious decisions:
- Single format (openai-json); children/zip is CLI-only. The HTTP
  endpoint and web UI are conversation-only, keeping the served surface
  — and its security surface (no child rows read through the coordinator
  handler) — small.
- `reasoning_content`, not the `reasoning` field /history and the
  reasoning-replay path use: export targets the chat-completions
  convention. Documented in export.py to prevent a "consistency fix".
- list_workstreams exposes no cursor, so the child walk passes an
  explicit high limit rather than inheriting the default 100, which
  would silently drop a coordinator's children past 100.
- Interactive export lives in the per-tab menu (interactive is
  per-tab/pane — avoids focused-workstream ambiguity); the coordinator
  is one conversation, so it keeps an appbar button.

Tested: 25 new tests through real storage + handlers (TestClient), incl.
cross-kind isolation 404, misconfig 500, the reasoning + attachment
pipeline, and the coordinator children zip. The shared frontend helper
is verified by a node sandbox harness (re-entrancy guard, button
disable/aria-busy, no-button tab-menu path). Full non-live suite green
(6714 passed); ruff + format + mypy clean; OpenAPI spec updated.
---
 tests/test_admin_export.py                    | 184 ++++++++++++++++
 tests/test_coordinator_endpoints.py           |  67 ++++++
 tests/test_export.py                          | 196 ++++++++++++++++++
 tests/test_workstream_endpoints.py            | 122 +++++++++++
 turnstone/admin.py                            |  39 ++++
 turnstone/api/console_spec.py                 |  14 ++
 turnstone/api/server_spec.py                  |  16 ++
 turnstone/console/server.py                   |   4 +
 .../console/static/coordinator/coordinator.js |  11 +
 .../console/static/coordinator/index.html     |   3 +
 turnstone/core/export.py                      | 119 +++++++++++
 turnstone/core/session_routes.py              | 131 ++++++++++++
 turnstone/server.py                           |   3 +
 turnstone/shared_static/utils.js              |  62 ++++++
 turnstone/ui/static/app.js                    |   6 +
 15 files changed, 977 insertions(+)
 create mode 100644 tests/test_admin_export.py
 create mode 100644 tests/test_export.py
 create mode 100644 turnstone/core/export.py

diff --git a/tests/test_admin_export.py b/tests/test_admin_export.py
new file mode 100644
index 00000000..5685fd5c
--- /dev/null
+++ b/tests/test_admin_export.py
@@ -0,0 +1,184 @@
+"""Tests for ``turnstone-admin export`` (issue #613, chunk 2).
+
+Drives the real ``_cmd_export`` handler through a real seeded SQLite DB
+(NOT a stubbed ``export_workstream``).  The handler builds its storage
+via ``_get_storage(args)``, so each test constructs an ``argparse.Namespace``
+whose DB attributes resolve to a tmp sqlite file, seeds that same file,
+then invokes the command.
+
+Seeding uses ``run_migrations=False`` (create_all builds the schema);
+``_cmd_export`` re-inits the same path with ``run_migrations=True`` (the
+admin default).  On SQLite the resulting "table already exists" Alembic
+error is swallowed as non-fatal, so the seeded rows survive — this mirrors
+the real CLI invocation path exactly.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import zipfile
+from typing import TYPE_CHECKING
+
+import pytest
+
+from turnstone.admin import _cmd_export
+from turnstone.core.storage import init_storage, reset_storage
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+
+@pytest.fixture(autouse=True)
+def _reset_storage_singleton() -> Iterator[None]:
+    """Keep the module-global storage singleton from leaking across tests."""
+    reset_storage()
+    yield
+    reset_storage()
+
+
+def _export_args(db_path: str, ws_id: str, *, children: bool, output: str) -> argparse.Namespace:
+    """Build the Namespace ``_cmd_export`` (via ``_get_storage``) expects.
+
+    ``_get_storage`` reads each DB field with ``getattr(args, name, None)``
+    and only falls back to the env var when the attribute ``is None``.
+    Pinning the string fields to ``""`` therefore short-circuits any
+    ``TURNSTONE_DB_*`` env leakage; ``db_backend``/``db_path`` point the
+    backend at the tmp sqlite file.
+    """
+    return argparse.Namespace(
+        ws_id=ws_id,
+        children=children,
+        output=output,
+        db_backend="sqlite",
+        db_path=db_path,
+        db_url="",
+        db_pool_size=2,
+        db_sslmode="",
+        db_sslrootcert="",
+        db_sslcert="",
+        db_sslkey="",
+    )
+
+
+def _seed_interactive(db_path: str, ws_id: str) -> list[str]:
+    """Seed one interactive workstream; return the seeded message roles in order."""
+    st = init_storage("sqlite", path=db_path, run_migrations=False)
+    st.register_workstream(ws_id, user_id="u1", title="Solo", kind="interactive")
+    roles = ["user", "assistant", "user", "assistant"]
+    st.save_message(ws_id, "user", "first question")
+    st.save_message(ws_id, "assistant", "first answer")
+    st.save_message(ws_id, "user", "second question")
+    st.save_message(ws_id, "assistant", "second answer")
+    return roles
+
+
+def _seed_coordinator(db_path: str, parent: str, children: list[str]) -> None:
+    """Seed a coordinator parent plus the given child workstreams."""
+    st = init_storage("sqlite", path=db_path, run_migrations=False)
+    st.register_workstream(parent, user_id="u1", title="Coord", kind="coordinator")
+    st.save_message(parent, "user", "coordinate")
+    st.save_message(parent, "assistant", "spawning children")
+    for child in children:
+        st.register_workstream(
+            child, user_id="u1", title=f"Child {child}", kind="interactive", parent_ws_id=parent
+        )
+        st.save_message(child, "user", "do work")
+        st.save_message(child, "assistant", "work done")
+
+
+def test_export_interactive_to_file(tmp_path: Path) -> None:
+    db_path = str(tmp_path / "admin.db")
+    seeded_roles = _seed_interactive(db_path, "ws_solo")
+    out_file = tmp_path / "out.json"
+
+    _cmd_export(_export_args(db_path, "ws_solo", children=False, output=str(out_file)))
+
+    payload = json.loads(out_file.read_bytes())
+    top_keys = sorted(payload.keys())
+    actual_roles = [m["role"] for m in payload["messages"]]
+    assert top_keys == ["messages"]
+    assert actual_roles == seeded_roles
+
+
+def test_export_to_stdout(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    db_path = str(tmp_path / "admin.db")
+    seeded_roles = _seed_interactive(db_path, "ws_solo")
+
+    _cmd_export(_export_args(db_path, "ws_solo", children=False, output="-"))
+
+    captured = capsys.readouterr()
+    payload = json.loads(captured.out)
+    has_messages = "messages" in payload
+    actual_roles = [m["role"] for m in payload["messages"]]
+    assert has_messages
+    assert actual_roles == seeded_roles
+
+
+def test_export_children_zip_to_file(tmp_path: Path) -> None:
+    db_path = str(tmp_path / "admin.db")
+    _seed_coordinator(db_path, "ws_parent", ["ws_kid_a", "ws_kid_b"])
+    out_file = tmp_path / "bundle.zip"
+
+    _cmd_export(_export_args(db_path, "ws_parent", children=True, output=str(out_file)))
+
+    with zipfile.ZipFile(out_file) as zf:
+        names = sorted(zf.namelist())
+    expected_names = [
+        "children/ws_kid_a.json",
+        "children/ws_kid_b.json",
+        "ws_parent.json",
+    ]
+    assert names == expected_names
+
+
+def test_export_unknown_ws_exits_1(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+    db_path = str(tmp_path / "admin.db")
+    # Seed an unrelated workstream so the DB/schema exist but the queried id does not.
+    _seed_interactive(db_path, "ws_present")
+
+    with pytest.raises(SystemExit) as exc_info:
+        _cmd_export(_export_args(db_path, "ws_absent", children=False, output="-"))
+
+    exit_code = exc_info.value.code
+    captured = capsys.readouterr()
+    stderr_has_not_found = "not found" in captured.err
+    assert exit_code == 1
+    assert stderr_has_not_found
+
+
+def test_export_children_zip_to_stdout(
+    tmp_path: Path, capsysbinary: pytest.CaptureFixture[bytes]
+) -> None:
+    db_path = str(tmp_path / "admin.db")
+    _seed_coordinator(db_path, "ws_parent", ["ws_kid_a"])
+
+    # Under pytest ``sys.stdout.isatty()`` is False, so the zip is written
+    # to ``sys.stdout.buffer`` as raw bytes (the pipe-friendly path).
+    _cmd_export(_export_args(db_path, "ws_parent", children=True, output="-"))
+
+    captured = capsysbinary.readouterr()
+    starts_with_zip_magic = captured.out.startswith(b"PK")
+    assert starts_with_zip_magic
+
+
+def test_export_children_zip_to_tty_refused(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    db_path = str(tmp_path / "admin.db")
+    _seed_coordinator(db_path, "ws_parent", ["ws_kid_a"])
+    # Force the "stdout is a terminal" branch: refuse to dump zip bytes.
+    monkeypatch.setattr(sys.stdout, "isatty", lambda: True, raising=False)
+
+    with pytest.raises(SystemExit) as exc_info:
+        _cmd_export(_export_args(db_path, "ws_parent", children=True, output="-"))
+
+    exit_code = exc_info.value.code
+    captured = capsys.readouterr()
+    stderr_has_refuse = "Refusing" in captured.err
+    assert exit_code == 1
+    assert stderr_has_refuse
diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py
index c4ad2837..e1d5e6c6 100644
--- a/tests/test_coordinator_endpoints.py
+++ b/tests/test_coordinator_endpoints.py
@@ -66,6 +66,7 @@ from turnstone.core.session_routes import (
     make_create_handler,
     make_dequeue_handler,
     make_detail_handler,
+    make_export_handler,
     make_history_handler,
     make_list_handler,
     make_open_handler,
@@ -200,6 +201,11 @@ def _make_client(
                 make_history_handler(_coord_endpoint_config),
                 methods=["GET"],
             ),
+            Route(
+                "/v1/api/workstreams/{ws_id}/export",
+                make_export_handler(_coord_endpoint_config),
+                methods=["GET"],
+            ),
             Route(
                 "/v1/api/workstreams/{ws_id}/open",
                 make_open_handler(_coord_endpoint_config),
@@ -1316,6 +1322,67 @@ def test_history_clamps_limit_query_param(storage):
     assert len(resp.json()["messages"]) == 6
 
 
+# ---------------------------------------------------------------------------
+# Export (issue #613) — conversation-only, never a zip
+# ---------------------------------------------------------------------------
+
+
+def test_export_happy_path_returns_json_not_zip(storage):
+    """A seeded coordinator exports as a JSON conversation envelope —
+    never a zip. The HTTP surface is conversation-only; the children/zip
+    capability is admin-CLI-only."""
+    mgr = _build_mgr(storage)
+    ws = mgr.create(user_id="user-1")
+    storage.save_message(ws.id, "user", "coordinate the work")
+    storage.save_message(ws.id, "assistant", "on it")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+
+    resp = client.get(f"/v1/api/workstreams/{ws.id}/export", headers=_COORD_HEADERS)
+    assert resp.status_code == 200
+    assert resp.headers["content-type"].startswith("application/json")
+    assert resp.headers["content-disposition"] == f'attachment; filename="{ws.id}.json"'
+    # NOT a zip — zip archives start with the "PK" local-file magic.
+    assert not resp.content.startswith(b"PK")
+    # Body parses to the OpenAI envelope with the seeded turns.
+    body = resp.json()
+    role_contents = [(m.get("role"), m.get("content")) for m in body["messages"]]
+    assert ("user", "coordinate the work") in role_contents
+
+
+def test_export_serves_storage_only_coordinator(storage):
+    """Closed / evicted coordinators export from storage without
+    rehydrating, same ladder history uses."""
+    mgr = _build_mgr(storage)
+    storage.register_workstream("storage-only-coord", kind="coordinator", user_id="user-1")
+    storage.save_message("storage-only-coord", "user", "from cold storage")
+    assert mgr.get("storage-only-coord") is None
+
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.get(
+        "/v1/api/workstreams/storage-only-coord/export",
+        headers=_COORD_HEADERS,
+    )
+    assert resp.status_code == 200
+    contents = [m.get("content") for m in resp.json()["messages"]]
+    assert "from cold storage" in contents
+    # Export does NOT rehydrate — pool stays cold.
+    assert mgr.get("storage-only-coord") is None
+
+
+def test_export_404_when_kind_interactive(storage):
+    """Cross-kind isolation: an interactive ws_id in shared storage 404s
+    on the coordinator export endpoint (the handler is built with
+    ``list_kind=COORDINATOR``). Proves the kind gate the same way
+    :func:`test_history_404_when_kind_interactive` does."""
+    mgr = _build_mgr(storage)
+    storage.register_workstream("ws-int", kind="interactive", user_id="user-1")
+    storage.save_message("ws-int", "user", "interactive content")
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+    resp = client.get("/v1/api/workstreams/ws-int/export", headers=_COORD_HEADERS)
+    assert resp.status_code == 404
+    assert "interactive content" not in resp.text
+
+
 # ---------------------------------------------------------------------------
 # Cancel
 # ---------------------------------------------------------------------------
diff --git a/tests/test_export.py b/tests/test_export.py
new file mode 100644
index 00000000..77e6511a
--- /dev/null
+++ b/tests/test_export.py
@@ -0,0 +1,196 @@
+"""Unit tests for the workstream export serializer (issue #613).
+
+Drives through a REAL storage backend (the ``backend`` fixture is a
+SQLite ``StorageBackend``): seed workstreams / messages / attachments,
+call :func:`export_workstream`, parse the returned bytes, and assert
+structural facts.  No hand-built message dicts are injected straight
+into the serializer as the sole gate — the pipeline order (attach
+reasoning → sanitize) is what these tests guard.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import zipfile
+
+from turnstone.core.export import (
+    WorkstreamNotFoundError,
+    _attach_reasoning_content,
+    _build_openai_json,
+    export_workstream,
+)
+
+USER = "u1"
+
+
+def _assistants(messages: list[dict]) -> list[dict]:
+    return [m for m in messages if m.get("role") == "assistant"]
+
+
+def _parse_messages(data: bytes) -> list[dict]:
+    return json.loads(data)["messages"]
+
+
+def _seed_interactive_turn(backend, ws_id: str) -> None:
+    """user + assistant(tool_call) + tool + assistant."""
+    tc = [
+        {
+            "id": "call_a1",
+            "type": "function",
+            "function": {"name": "run", "arguments": "{}"},
+        }
+    ]
+    backend.register_workstream(ws_id, user_id=USER, title="T", kind="interactive")
+    backend.save_message(ws_id, "user", "go")
+    backend.save_message(ws_id, "assistant", "working", tool_calls=json.dumps(tc))
+    backend.save_message(ws_id, "tool", "ran ok", tool_name="run", tool_call_id="call_a1")
+    backend.save_message(ws_id, "assistant", "done")
+
+
+def test_openai_json_envelope_shape(backend):
+    _seed_interactive_turn(backend, "ws1")
+    result = export_workstream(backend, "ws1")
+
+    assert result.content_type == "application/json"
+    assert result.filename == "ws1.json"
+    top_keys = sorted(json.loads(result.data).keys())
+    assert top_keys == ["messages"]
+
+
+def test_reasoning_content_present_thinking(backend):
+    pc = [{"type": "thinking", "thinking": "R1", "signature": "sig"}]
+    backend.register_workstream("ws1", user_id=USER, kind="interactive")
+    backend.save_message("ws1", "user", "go")
+    backend.save_message("ws1", "assistant", "ok", provider_data=json.dumps(pc))
+
+    messages = _parse_messages(export_workstream(backend, "ws1").data)
+    reasoning = [m.get("reasoning_content") for m in _assistants(messages)]
+    assert reasoning == ["R1"]
+
+
+def test_reasoning_content_present_reasoning_text(backend):
+    pc = [{"type": "reasoning_text", "text": "R2", "source": "synth"}]
+    backend.register_workstream("ws1", user_id=USER, kind="interactive")
+    backend.save_message("ws1", "user", "go")
+    backend.save_message("ws1", "assistant", "ok", provider_data=json.dumps(pc))
+
+    messages = _parse_messages(export_workstream(backend, "ws1").data)
+    reasoning = [m.get("reasoning_content") for m in _assistants(messages)]
+    assert reasoning == ["R2"]
+
+
+def test_reasoning_content_present_responses(backend):
+    pc = [{"type": "reasoning", "summary": [{"type": "summary_text", "text": "R3"}]}]
+    backend.register_workstream("ws1", user_id=USER, kind="interactive")
+    backend.save_message("ws1", "user", "go")
+    backend.save_message("ws1", "assistant", "ok", provider_data=json.dumps(pc))
+
+    messages = _parse_messages(export_workstream(backend, "ws1").data)
+    reasoning_text = _assistants(messages)[0].get("reasoning_content")
+    assert reasoning_text is not None
+    assert "R3" in reasoning_text
+
+
+def test_no_underscore_keys_leak(backend):
+    pc = [{"type": "thinking", "thinking": "R1", "signature": "sig"}]
+    _seed_interactive_turn(backend, "ws1")
+    backend.save_message("ws1", "assistant", "more", provider_data=json.dumps(pc))
+
+    messages = _parse_messages(export_workstream(backend, "ws1").data)
+    leaked = sorted({k for m in messages for k in m if isinstance(k, str) and k.startswith("_")})
+    assert leaked == []
+
+
+def test_image_url_kept_document_inlined(backend):
+    backend.register_workstream("ws1", user_id=USER, kind="interactive")
+    msg_id = backend.save_message("ws1", "user", "see attached")
+    backend.save_attachment("att_img", "ws1", USER, "pic.png", "image/png", 4, "image", b"\x89PNG")
+    backend.save_attachment("att_doc", "ws1", USER, "notes.txt", "text/plain", 5, "text", b"hello")
+    backend.mark_attachments_consumed(["att_img", "att_doc"], msg_id, "ws1", USER)
+    backend.save_message("ws1", "assistant", "got it")
+
+    messages = _parse_messages(export_workstream(backend, "ws1").data)
+    user_msg = next(m for m in messages if m.get("role") == "user")
+    parts = user_msg["content"]
+    part_types = [p.get("type") for p in parts]
+    document_texts = [
+        p.get("text", "")
+        for p in parts
+        if p.get("type") == "text" and "<document name=" in p.get("text", "")
+    ]
+
+    assert "image_url" in part_types
+    assert document_texts != []
+
+
+def test_assistant_without_reasoning_has_no_reasoning_content(backend):
+    backend.register_workstream("ws1", user_id=USER, kind="interactive")
+    backend.save_message("ws1", "user", "go")
+    backend.save_message("ws1", "assistant", "ok")
+
+    messages = _parse_messages(export_workstream(backend, "ws1").data)
+    assistant = _assistants(messages)[0]
+    assert "reasoning_content" not in assistant
+
+
+def test_coordinator_zip_parent_plus_children(backend):
+    backend.register_workstream("coord", user_id=USER, title="C", kind="coordinator")
+    backend.save_message("coord", "user", "coordinate")
+    backend.save_message("coord", "assistant", "spawning")
+    backend.register_workstream("c1", user_id=USER, kind="interactive", parent_ws_id="coord")
+    backend.register_workstream("c2", user_id=USER, kind="interactive", parent_ws_id="coord")
+    for child in ("c1", "c2"):
+        backend.save_message(child, "user", "do x")
+        backend.save_message(child, "assistant", "x done")
+
+    result = export_workstream(backend, "coord", children=True)
+    assert result.content_type == "application/zip"
+    assert result.filename == "coord.zip"
+
+    zf = zipfile.ZipFile(io.BytesIO(result.data))
+    names = sorted(zf.namelist())
+    expected = sorted(["coord.json", "children/c1.json", "children/c2.json"])
+    assert names == expected
+
+    top_keys = [sorted(json.loads(zf.read(name)).keys()) for name in names]
+    assert top_keys == [["messages"], ["messages"], ["messages"]]
+
+
+def test_coordinator_default_parent_only(backend):
+    backend.register_workstream("coord", user_id=USER, title="C", kind="coordinator")
+    backend.save_message("coord", "user", "coordinate")
+    backend.save_message("coord", "assistant", "done")
+    backend.register_workstream("c1", user_id=USER, kind="interactive", parent_ws_id="coord")
+
+    result = export_workstream(backend, "coord", children=False)
+    assert result.content_type == "application/json"
+    assert result.filename == "coord.json"
+
+
+def test_export_unknown_ws_raises(backend):
+    try:
+        export_workstream(backend, "does-not-exist")
+    except WorkstreamNotFoundError as exc:
+        assert "does-not-exist" in str(exc)
+    else:
+        raise AssertionError("expected WorkstreamNotFoundError")
+
+
+def test_attach_reasoning_runs_before_sanitize(backend):
+    pc = [{"type": "thinking", "thinking": "R1", "signature": "sig"}]
+    backend.register_workstream("ws1", user_id=USER, kind="interactive")
+    backend.save_message("ws1", "user", "go")
+    backend.save_message("ws1", "assistant", "ok", provider_data=json.dumps(pc))
+
+    attached = _attach_reasoning_content(backend.load_messages("ws1", repair=True))
+    assistant = _assistants(attached)[0]
+    # Pre-sanitize: reasoning stamped AND the raw provider lane still present.
+    assert assistant.get("reasoning_content") == "R1"
+    assert "_provider_content" in assistant
+
+    # Full pipeline output: provider lane is gone, reasoning survives.
+    messages = _parse_messages(_build_openai_json(backend, "ws1"))
+    leaked = [k for m in messages for k in m if isinstance(k, str) and k.startswith("_")]
+    assert leaked == []
+    assert _assistants(messages)[0].get("reasoning_content") == "R1"
diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py
index b70c776c..178401df 100644
--- a/tests/test_workstream_endpoints.py
+++ b/tests/test_workstream_endpoints.py
@@ -28,6 +28,7 @@ from turnstone.core.history_decoration import (
 from turnstone.core.session_routes import (
     SessionEndpointConfig,
     make_detail_handler,
+    make_export_handler,
     make_history_handler,
     make_open_handler,
     make_retry_handler,
@@ -983,6 +984,127 @@ def _build_detail_app(
     return TestClient(app)
 
 
+def _build_export_app(
+    mock_mgr: Any,
+    storage: Any,
+    *,
+    cfg: SessionEndpointConfig | None = None,
+) -> TestClient:
+    """Mount the lifted ``export`` factory at ``/{ws_id}/export``.
+
+    Mirrors :func:`_build_history_app` — real factory, real storage on
+    ``app.state.auth_storage``, driven via ``TestClient``. The optional
+    ``cfg`` override lets the misconfig / cross-kind tests swap in a cfg
+    with a deliberately wrong (or ``None``) ``list_kind``.
+    """
+    if cfg is None:
+        cfg = _interactive_endpoint_cfg(mock_mgr)
+    handler = make_export_handler(cfg)
+    app = Starlette(
+        routes=[
+            Mount(
+                "/v1",
+                routes=[
+                    Route("/api/workstreams/{ws_id}/export", handler, methods=["GET"]),
+                ],
+            ),
+        ],
+        middleware=[Middleware(_InjectAuthMiddleware)],
+    )
+    app.state.workstreams = mock_mgr
+    app.state.auth_storage = storage
+    return TestClient(app)
+
+
+class TestExportInteractive:
+    """Interactive coverage for the lifted, conversation-only
+    ``GET /v1/api/workstreams/{ws_id}/export`` (issue #613)."""
+
+    def test_happy_path_returns_json_download(self, _inject_storage):
+        ws_id = "ws-export-1"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        _inject_storage.save_message(ws_id, "user", "export me")
+        _inject_storage.save_message(ws_id, "assistant", "exported")
+        mock_ws = MagicMock()
+        mock_ws.id = ws_id
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = mock_ws
+        client = _build_export_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/export")
+        assert r.status_code == 200
+        assert r.headers["content-type"].startswith("application/json")
+        assert r.headers["content-disposition"] == f'attachment; filename="{ws_id}.json"'
+        assert r.headers["x-content-type-options"] == "nosniff"
+        # Parse the actual bytes — conversation envelope with the seeded turns.
+        body = json.loads(r.content)
+        role_contents = [(m.get("role"), m.get("content")) for m in body["messages"]]
+        assert "messages" in body
+        assert ("user", "export me") in role_contents
+
+    def test_serves_storage_only_workstream(self, _inject_storage):
+        """A persisted-but-not-loaded interactive exports without
+        rehydrating — same storage-fallback ladder history uses."""
+        ws_id = "ws-export-cold"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        _inject_storage.save_message(ws_id, "assistant", "from cold storage")
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = None  # not loaded
+        client = _build_export_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/export")
+        assert r.status_code == 200
+        body = json.loads(r.content)
+        contents = [m.get("content") for m in body["messages"]]
+        assert "from cold storage" in contents
+
+    def test_404_on_missing_ws_id(self, _inject_storage):
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = None
+        client = _build_export_app(mock_mgr, _inject_storage)
+
+        r = client.get("/v1/api/workstreams/no-such-ws/export")
+        assert r.status_code == 404
+        assert r.json()["error"] == "Workstream not found"
+
+    def test_404_on_cross_kind_coord_ws_id(self, _inject_storage):
+        """Cross-kind isolation on the storage fallback: a coord ws_id in
+        shared storage 404s on the interactive export endpoint."""
+        ws_id = "ws-export-coord"
+        _inject_storage.register_workstream(ws_id, kind="coordinator", user_id="test-user")
+        _inject_storage.save_message(ws_id, "user", "coord-only content")
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = None
+        client = _build_export_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/export")
+        assert r.status_code == 404
+        assert "coord-only content" not in r.text
+
+    def test_500_when_list_kind_misconfigured(self, _inject_storage):
+        """A cfg mounted without ``list_kind`` fails loud (500) rather
+        than leaking cross-kind rows through the storage fallback."""
+        ws_id = "ws-export-misconfig"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        _inject_storage.save_message(ws_id, "user", "should not leak")
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = None
+        bad_cfg = SessionEndpointConfig(
+            permission_gate=None,
+            manager_lookup=lambda _r: (mock_mgr, None),
+            tenant_check=None,
+            not_found_label="Workstream not found",
+            audit_action_prefix="workstream",
+            list_kind=None,  # deliberately unset → fail loud
+        )
+        client = _build_export_app(mock_mgr, _inject_storage, cfg=bad_cfg)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/export")
+        assert r.status_code == 500
+        assert r.json()["error"] == "export handler misconfigured"
+        assert "should not leak" not in r.text
+
+
 class TestHistoryInteractive:
     """Interactive parity for the lifted ``GET /v1/api/workstreams/{ws_id}/history``."""
 
diff --git a/turnstone/admin.py b/turnstone/admin.py
index 3362efc7..109998f3 100644
--- a/turnstone/admin.py
+++ b/turnstone/admin.py
@@ -383,6 +383,34 @@ def _cmd_delete_node_metadata(args: argparse.Namespace) -> None:
         sys.exit(1)
 
 
+def _cmd_export(args: argparse.Namespace) -> None:
+    """Export a workstream as an OpenAI messages envelope (JSON, or zip with --children)."""
+    from turnstone.core.export import WorkstreamNotFoundError, export_workstream
+
+    storage = _get_storage(args)
+    try:
+        result = export_workstream(storage, args.ws_id, children=args.children)
+    except WorkstreamNotFoundError:
+        print(f"Workstream not found: {args.ws_id}", file=sys.stderr)
+        sys.exit(1)
+
+    if args.output == "-":
+        if result.content_type == "application/zip" and sys.stdout.isatty():
+            print(
+                "Refusing to write zip bytes to a terminal; use --output FILE or pipe.",
+                file=sys.stderr,
+            )
+            sys.exit(1)
+        if result.content_type == "application/zip":
+            sys.stdout.buffer.write(result.data)
+        else:
+            sys.stdout.write(result.data.decode("utf-8"))
+    else:
+        with open(args.output, "wb") as fh:
+            fh.write(result.data)
+        print(f"Wrote {len(result.data)} bytes to {args.output}", file=sys.stderr)
+
+
 def _discover_console_url() -> str:
     """Discover console URL from the services table."""
     from turnstone.core.storage import get_storage
@@ -485,6 +513,16 @@ def main() -> None:
     p_dnm.add_argument("node_id", help="Node ID")
     p_dnm.add_argument("key", help="Metadata key")
 
+    # Export
+    p_export = sub.add_parser("export", help="Export a workstream as OpenAI messages JSON")
+    p_export.add_argument("ws_id", help="Workstream id to export")
+    p_export.add_argument(
+        "--children",
+        action="store_true",
+        help="Bundle the coordinator parent + one JSON per child as a zip",
+    )
+    p_export.add_argument("--output", "-o", default="-", help="Output file path, or - for stdout")
+
     args = parser.parse_args()
     if not args.command:
         parser.print_help()
@@ -503,5 +541,6 @@ def main() -> None:
         "list-node-metadata": _cmd_list_node_metadata,
         "set-node-metadata": _cmd_set_node_metadata,
         "delete-node-metadata": _cmd_delete_node_metadata,
+        "export": _cmd_export,
     }
     dispatch[args.command](args)
diff --git a/turnstone/api/console_spec.py b/turnstone/api/console_spec.py
index 508eca54..4545903a 100644
--- a/turnstone/api/console_spec.py
+++ b/turnstone/api/console_spec.py
@@ -1415,6 +1415,20 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [
         error_codes=[400, 403, 404, 500, 503],
         tags=["Coordinator"],
     ),
+    EndpointSpec(
+        "/v1/api/workstreams/{ws_id}/export",
+        "GET",
+        "Export the coordinator's conversation as OpenAI messages JSON",
+        description=(
+            "Returns the coordinator's own conversation as an "
+            '``{"messages": [...]}`` OpenAI Chat Completions envelope, '
+            "served as a ``<ws_id>.json`` file download. Conversation-only "
+            "(children are not bundled over HTTP). Gated on "
+            "``admin.coordinator``."
+        ),
+        error_codes=[400, 403, 404, 500, 503],
+        tags=["Coordinator"],
+    ),
     EndpointSpec(
         "/v1/api/workstreams/{ws_id}/children",
         "GET",
diff --git a/turnstone/api/server_spec.py b/turnstone/api/server_spec.py
index 004ad9b2..d7107c94 100644
--- a/turnstone/api/server_spec.py
+++ b/turnstone/api/server_spec.py
@@ -258,6 +258,22 @@ SERVER_ENDPOINTS: list[EndpointSpec] = [
         error_codes=[400, 404, 500, 503],
         tags=["Workstreams"],
     ),
+    EndpointSpec(
+        "/v1/api/workstreams/{ws_id}/export",
+        "GET",
+        "Export the workstream's conversation as OpenAI messages JSON",
+        description=(
+            'Returns the full conversation as an ``{"messages": [...]}`` '
+            "OpenAI Chat Completions envelope, served as a ``<ws_id>.json`` "
+            "file download (``Content-Disposition: attachment``). Persisted "
+            "reasoning is surfaced on assistant messages as a "
+            "``reasoning_content`` field. Conversation-only — the parent + "
+            "per-child zip bundle is exposed only through the "
+            "``turnstone-admin export --children`` CLI."
+        ),
+        error_codes=[400, 404, 500, 503],
+        tags=["Workstreams"],
+    ),
     # --- Workstream attachments ---
     EndpointSpec(
         "/v1/api/workstreams/{ws_id}/attachments",
diff --git a/turnstone/console/server.py b/turnstone/console/server.py
index e0323910..570c5b0e 100644
--- a/turnstone/console/server.py
+++ b/turnstone/console/server.py
@@ -69,6 +69,7 @@ from turnstone.core.session_routes import (
     make_dequeue_handler,
     make_detail_handler,
     make_events_handler,
+    make_export_handler,
     make_history_handler,
     make_list_handler,
     make_open_handler,
@@ -12660,6 +12661,9 @@ def create_app(
             ),
             events=make_events_handler(coord_endpoint_config),  # lifted: shared body
             history=make_history_handler(coord_endpoint_config),  # lifted: shared body
+            export=make_export_handler(  # lifted: shared body (#613, conversation-only)
+                coord_endpoint_config
+            ),
             attachments=make_attachment_handlers(
                 coord_endpoint_config
             ),  # lifted: shared body (P1.5)
diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js
index dedfa006..c6188e89 100644
--- a/turnstone/console/static/coordinator/coordinator.js
+++ b/turnstone/console/static/coordinator/coordinator.js
@@ -4076,6 +4076,17 @@
     });
   }
 
+  // Appbar Export button — download this coordinator's conversation as
+  // OpenAI-shaped JSON via the shared helper in utils.js.  Wired here
+  // (rather than via an inline onclick like coord-close-btn) because the
+  // helper needs the IIFE-scoped ``wsId`` const captured at load time.
+  const exportBtn = document.getElementById("coord-export-btn");
+  if (exportBtn) {
+    exportBtn.addEventListener("click", () => {
+      exportWorkstreamDownload(wsId, exportBtn);
+    });
+  }
+
   // Mobile-only sidebar toggle — wires the accordion collapse below 700px.
   // On desktop the button is display:none so the handler is a no-op.
   const sidebarEl = document.getElementById("coord-sidebar");
diff --git a/turnstone/console/static/coordinator/index.html b/turnstone/console/static/coordinator/index.html
index 5bbc4985..0e4bfe14 100644
--- a/turnstone/console/static/coordinator/index.html
+++ b/turnstone/console/static/coordinator/index.html
@@ -532,6 +532,7 @@
   <span class="appbar-spacer"></span>
   <span id="coord-sse-status" class="appbar-status" aria-live="polite">connecting…</span>
   <span class="appbar-actions">
+    <button id="coord-export-btn" class="btn" type="button" aria-label="Export conversation" title="Export conversation (OpenAI JSON)">&#10515;</button>
     <button id="coord-close-btn" class="btn" onclick="coordCloseSession()" title="End this coordinator session (terminates it on the server)">end</button>
     <button id="theme-toggle" class="btn" onclick="toggleTheme()" aria-label="Toggle light/dark theme">&#9790;</button>
   </span>
@@ -618,6 +619,8 @@
   </aside>
 </div>
 
+<div id="toast" role="status" aria-live="polite"></div>
+
 <!-- Shared-static import order matches the console dashboard
      (console/static/index.html) so global shortcuts (kb.js) and helpers
      register in the expected sequence.  The renderer libs (katex, hljs,
diff --git a/turnstone/core/export.py b/turnstone/core/export.py
new file mode 100644
index 00000000..2b02fded
--- /dev/null
+++ b/turnstone/core/export.py
@@ -0,0 +1,119 @@
+"""Workstream export serializer (issue #613).
+
+Pure transform from a storage handle to an OpenAI-style conversation
+envelope (``{"messages": [...]}``).  The storage handle is injected by
+the caller — this module NEVER imports a storage singleton — so the
+admin CLI and the HTTP handler share one serializer.
+
+A single workstream exports as JSON bytes; a coordinator exported with
+``children=True`` exports as a zip bundling the parent at ``<ws_id>.json``
+and each child at ``children/<child_id>.json`` (no manifest).
+
+The reasoning lane is surfaced before sanitization: each assistant
+message's stored ``_provider_content`` is run through the pure
+:func:`extract_reasoning_text_from_provider_content` primitive and, when
+non-empty, stamped onto a ``reasoning_content`` field.  That ordering is
+load-bearing — :func:`sanitize_messages` strips every ``_``-prefixed key,
+so reasoning must be lifted out of ``_provider_content`` first.  The
+``reasoning_content`` key follows the chat-completions convention
+(vLLM / DeepSeek) and is deliberately distinct from the ``/history`` UI
+surface, which stamps the same text onto a ``reasoning`` field
+(see ``history_decoration.extract_reasoning_for_history``).
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import zipfile
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from turnstone.core.history_decoration import (
+    extract_reasoning_text_from_provider_content,
+)
+from turnstone.core.providers._openai_common import sanitize_messages
+
+if TYPE_CHECKING:
+    from turnstone.core.storage import StorageBackend
+
+# ``list_workstreams`` defaults to limit=100 and exposes no offset/cursor,
+# so the child walk passes an effectively-unbounded ceiling to avoid
+# silently truncating a coordinator's children from the export.
+_CHILD_EXPORT_LIMIT = 1_000_000
+
+
+@dataclass(frozen=True)
+class ExportResult:
+    """The serialized export and the HTTP metadata for serving it."""
+
+    data: bytes
+    content_type: str
+    filename: str
+
+
+class WorkstreamNotFoundError(Exception):
+    """Raised by :func:`export_workstream` when ``ws_id`` has no row."""
+
+
+def _attach_reasoning_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Stamp ``reasoning_content`` on assistant messages from stored reasoning.
+
+    Returns a NEW list.  For each assistant message, the stored
+    ``_provider_content`` is dispatched through
+    :func:`extract_reasoning_text_from_provider_content`; when it yields
+    non-empty text a new dict carrying ``reasoning_content`` is emitted,
+    otherwise the message passes through unchanged.  Must run BEFORE
+    :func:`sanitize_messages`, which strips ``_provider_content``.
+    """
+    out: list[dict[str, Any]] = []
+    for msg in messages:
+        if msg.get("role") == "assistant":
+            text = extract_reasoning_text_from_provider_content(msg.get("_provider_content"))
+            if text:
+                out.append({**msg, "reasoning_content": text})
+                continue
+        out.append(msg)
+    return out
+
+
+def _build_openai_json(storage: StorageBackend, ws_id: str) -> bytes:
+    """Serialize one workstream's history as an OpenAI envelope (JSON bytes)."""
+    messages = sanitize_messages(
+        _attach_reasoning_content(storage.load_messages(ws_id, repair=True))
+    )
+    return json.dumps({"messages": messages}, ensure_ascii=False, indent=2).encode("utf-8")
+
+
+def _list_child_ws_ids(storage: StorageBackend, ws_id: str) -> list[str]:
+    """Return the ws_ids of every workstream whose parent is ``ws_id``."""
+    return [
+        r._mapping["ws_id"]
+        for r in storage.list_workstreams(parent_ws_id=ws_id, limit=_CHILD_EXPORT_LIMIT)
+    ]
+
+
+def export_workstream(
+    storage: StorageBackend, ws_id: str, *, children: bool = False
+) -> ExportResult:
+    """Export a workstream as an OpenAI-style conversation envelope.
+
+    With ``children=False`` (default) the result is JSON bytes for the
+    single workstream.  With ``children=True`` the result is a zip
+    bundling the parent and each child workstream (one JSON entry each).
+
+    Raises :class:`WorkstreamNotFoundError` when ``ws_id`` has no row.
+    """
+    if storage.get_workstream(ws_id) is None:
+        raise WorkstreamNotFoundError(ws_id)
+
+    parent_bytes = _build_openai_json(storage, ws_id)
+    if not children:
+        return ExportResult(parent_bytes, "application/json", f"{ws_id}.json")
+
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
+        zf.writestr(f"{ws_id}.json", parent_bytes)
+        for child_id in _list_child_ws_ids(storage, ws_id):
+            zf.writestr(f"children/{child_id}.json", _build_openai_json(storage, child_id))
+    return ExportResult(buf.getvalue(), "application/zip", f"{ws_id}.zip")
diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py
index d73853c3..9b84ad8a 100644
--- a/turnstone/core/session_routes.py
+++ b/turnstone/core/session_routes.py
@@ -522,6 +522,7 @@ class SharedSessionVerbHandlers:
     retry: Handler | None = None  # POST {prefix}/{ws_id}/retry
     events: Handler | None = None  # GET  {prefix}/{ws_id}/events (SSE)
     history: Handler | None = None  # GET  {prefix}/{ws_id}/history
+    export: Handler | None = None  # GET  {prefix}/{ws_id}/export
 
     # Attachments — the four handlers come together or not at all.
     attachments: AttachmentHandlers | None = None
@@ -615,6 +616,8 @@ def register_session_routes(
         routes.append(Route(f"{p}/{{ws_id}}/events", handlers.events, methods=["GET"]))
     if handlers.history is not None:
         routes.append(Route(f"{p}/{{ws_id}}/history", handlers.history, methods=["GET"]))
+    if handlers.export is not None:
+        routes.append(Route(f"{p}/{{ws_id}}/export", handlers.export, methods=["GET"]))
 
     # --- Attachments (the quartet comes together or not at all) ---------
     if handlers.attachments is not None:
@@ -2924,6 +2927,134 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
     return history
 
 
+def make_export_handler(cfg: SessionEndpointConfig) -> Handler:
+    """Lifted body for ``GET {prefix}/{ws_id}/export`` — conversation download.
+
+    Serves the workstream's full conversation as an OpenAI-style
+    envelope (``{"messages": [...]}``) for download. Reuses the same
+    :class:`SessionEndpointConfig` (and therefore the same gate ladder)
+    as :func:`make_history_handler`, so ownership + cross-kind isolation
+    come for free.
+
+    The HTTP surface is **conversation-only**: it never bundles
+    children and always returns ``application/json``. The
+    children/zip capability of :func:`export_workstream` is reserved
+    for the admin CLI, so the handler calls it with the default
+    ``children=False``.
+
+    Per-kind divergence captured by the same fields history consults:
+
+    - ``cfg.permission_gate`` — coord's ``admin.coordinator`` check;
+      interactive ``None``.
+    - ``cfg.manager_lookup`` — the kind's manager.
+    - ``cfg.list_kind`` — required for the storage-fallback kind check
+      so an interactive ws_id can't be exported through the coord
+      process and vice versa. **Required when this handler is
+      mounted** — a missing value fails loud (500 + ``log.error``)
+      rather than silently leaking cross-kind history.
+    - ``cfg.tenant_check`` — per-``ws_id`` access gate.
+    - ``cfg.not_found_label`` — per-kind 404 wording.
+
+    Args:
+        cfg: per-kind policy bundle.
+    """
+
+    async def export(request: Request) -> Response:
+        import asyncio
+
+        if cfg.permission_gate is not None:
+            err = cfg.permission_gate(request)
+            if err is not None:
+                return err
+
+        # Fail-closed misconfig gate. Without ``cfg.list_kind`` the
+        # storage-fallback path below has no way to enforce cross-kind
+        # isolation — an interactive ws_id requested through a coord
+        # process would silently export coord history from storage (and
+        # vice versa). Mirrors :func:`make_history_handler`'s same gate.
+        if cfg.list_kind is None:
+            log.error("ws.export.misconfigured_no_list_kind")
+            return JSONResponse(
+                {"error": "export handler misconfigured"},
+                status_code=500,
+            )
+
+        mgr_opt, err503 = cfg.manager_lookup(request)
+        if err503 is not None:
+            return err503
+        mgr = cast("SessionManager", mgr_opt)
+
+        ws_id = request.path_params.get("ws_id", "")
+        if not ws_id:
+            return JSONResponse({"error": "ws_id is required"}, status_code=400)
+
+        # Cross-tenant gate — same posture as history (interactive wires
+        # ``_interactive_tenant_check``; coord wires ``None`` and relies
+        # on the ``admin.coordinator`` permission_gate above). Always
+        # offloaded via ``to_thread`` since the interactive resolver
+        # falls through to a synchronous storage read on a cache miss.
+        if cfg.tenant_check is not None:
+            err_tenant = await asyncio.to_thread(cfg.tenant_check, request, ws_id, mgr)
+            if err_tenant is not None:
+                return err_tenant
+
+        # Existence + kind check. The workstream may live only in
+        # storage (closed coordinators / persisted-but-not-loaded
+        # interactives are still exportable without rehydrating).
+        # Mirrors history's ladder: in-memory mgr.get → storage row +
+        # kind check → 404. Falling back to storage without the kind
+        # check would leak interactive rows through the coord endpoint
+        # (and vice versa) on a process that shares storage with the
+        # other kind. ``cfg.list_kind`` is guaranteed non-None above.
+        storage = getattr(request.app.state, "auth_storage", None)
+        live_session = mgr.get(ws_id)
+        if live_session is None:
+            if storage is None:
+                return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+            try:
+                row = await asyncio.to_thread(storage.get_workstream, ws_id)
+            except Exception:
+                log.debug("ws.export.lookup_failed ws=%s", ws_id[:8], exc_info=True)
+                return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+            if row is None or row.get("kind") != cfg.list_kind:
+                return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+
+        # Past the gate but no storage handle — the live-session branch
+        # above skips the storage requirement, but the serializer needs
+        # a real handle. Degrade to the same 404 the fallback uses for a
+        # missing storage rather than serving an empty / 500 export.
+        if storage is None:
+            return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+
+        from starlette.responses import Response as _Response
+
+        from turnstone.core.export import WorkstreamNotFoundError, export_workstream
+
+        # Conversation-only: never bundle children, always JSON.  A live
+        # session whose storage row was deleted skips the fallback
+        # existence gate above, so guard the serializer's own not-found
+        # raise and degrade to the same 404 rather than surfacing a 500.
+        try:
+            result = await asyncio.to_thread(export_workstream, storage, ws_id)
+        except WorkstreamNotFoundError:
+            return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+        # ws_ids are hex so the filename is already safe, but mirror the
+        # attachment download handler's defensive strip of quotes/CR/LF
+        # so a future non-hex id can't break the Content-Disposition.
+        safe_name = result.filename.replace('"', "").replace("\r", "").replace("\n", "")
+        return _Response(
+            result.data,
+            media_type=result.content_type,
+            headers={
+                "Content-Disposition": f'attachment; filename="{safe_name}"',
+                "X-Content-Type-Options": "nosniff",
+                "Cache-Control": "private, no-store",
+            },
+        )
+
+    return export
+
+
 def make_detail_handler(cfg: SessionEndpointConfig) -> Handler:
     """Lifted body for ``GET {prefix}/{ws_id}`` — workstream display fields.
 
diff --git a/turnstone/server.py b/turnstone/server.py
index 404dff76..bd9b7adb 100644
--- a/turnstone/server.py
+++ b/turnstone/server.py
@@ -72,6 +72,7 @@ from turnstone.core.session_routes import (
     make_dequeue_handler,
     make_detail_handler,
     make_events_handler,
+    make_export_handler,
     make_history_handler,
     make_list_handler,
     make_open_handler,
@@ -3798,6 +3799,7 @@ def create_app(
     list_handler = make_list_handler(interactive_endpoint_config)
     saved_handler = make_saved_handler(interactive_endpoint_config)
     history_handler = make_history_handler(interactive_endpoint_config)
+    export_handler = make_export_handler(interactive_endpoint_config)
     detail_handler = make_detail_handler(interactive_endpoint_config)
     v1_routes: list[Any] = [
         Route("/api/events/global", global_events_sse),
@@ -3823,6 +3825,7 @@ def create_app(
             retry=retry_handler,  # lifted: shared body (#549)
             events=events_handler,  # lifted: shared body
             history=history_handler,  # lifted: shared body (interactive feature gain)
+            export=export_handler,  # lifted: shared body (#613, conversation-only)
             attachments=attachment_handlers,  # lifted: shared body (P1.5)
         ),
     )
diff --git a/turnstone/shared_static/utils.js b/turnstone/shared_static/utils.js
index cfcc9247..69228ab6 100644
--- a/turnstone/shared_static/utils.js
+++ b/turnstone/shared_static/utils.js
@@ -140,3 +140,65 @@ function replayAdvisoriesAfterTool(advisories, renderUserText) {
     renderUserText(adv.text || "");
   }
 }
+
+// Download a workstream's conversation as OpenAI-shaped JSON.  Hits
+// GET /v1/api/workstreams/{ws_id}/export, which streams a
+// ``{"messages":[...]}`` body with a Content-Disposition attachment
+// filename.  Shared by the interactive appbar (app.js) and the
+// coordinator appbar (coordinator.js) so both export buttons behave
+// identically.  authFetch already handles the 401 (shows login) and
+// 429 (retry) paths and returns the raw Response, so we read .blob()
+// directly and synthesise an anchor click to trigger the browser save.
+async function exportWorkstreamDownload(wsId, btn) {
+  if (!wsId) {
+    showToast("No conversation to export", "error");
+    return;
+  }
+  // Re-entrancy guard: a double-click (or Enter+Enter) must not fire two
+  // concurrent exports / two downloads.  The optional triggering button
+  // is disabled for the duration as the in-progress affordance, matching
+  // the send/stop buttons' disable-during-async pattern.
+  if (exportWorkstreamDownload._busy) return;
+  exportWorkstreamDownload._busy = true;
+  if (btn) {
+    btn.disabled = true;
+    btn.setAttribute("aria-busy", "true");
+  }
+  try {
+    const url = "/v1/api/workstreams/" + encodeURIComponent(wsId) + "/export";
+    let r;
+    try {
+      r = await authFetch(url);
+    } catch (e) {
+      // authFetch throws Error("auth") on 401 after showing the login
+      // modal — nothing more to do here.
+      return;
+    }
+    if (!r || !r.ok) {
+      showToast("Export failed", "error");
+      return;
+    }
+    let filename = wsId + ".json";
+    const cd = r.headers.get("Content-Disposition");
+    if (cd) {
+      const m = cd.match(/filename="([^"]+)"/);
+      if (m) filename = m[1];
+    }
+    const blob = await r.blob();
+    const objUrl = URL.createObjectURL(blob);
+    const a = document.createElement("a");
+    a.href = objUrl;
+    a.download = filename;
+    document.body.appendChild(a);
+    a.click();
+    a.remove();
+    URL.revokeObjectURL(objUrl);
+    showToast("Exported " + filename);
+  } finally {
+    exportWorkstreamDownload._busy = false;
+    if (btn) {
+      btn.disabled = false;
+      btn.removeAttribute("aria-busy");
+    }
+  }
+}
diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js
index 410f2396..10a37e00 100644
--- a/turnstone/ui/static/app.js
+++ b/turnstone/ui/static/app.js
@@ -2994,6 +2994,12 @@ function showTabDropdown(chevronEl, wsId) {
         forkWorkstream(wsId);
       },
     },
+    {
+      label: "Export conversation",
+      action: function () {
+        exportWorkstreamDownload(wsId);
+      },
+    },
     {
       label: "Close",
       key: "Ctrl+W",