mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
feat(export): export workstream conversations as OpenAI messages JSON
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.
This commit is contained in:
Patrick Buckley
@@ -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
|
||||
@@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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"
|
||||
@@ -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``."""
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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)">⤓</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">☾</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,
|
||||
|
||||
@@ -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")
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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)
|
||||
),
|
||||
)
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2994,6 +2994,12 @@ function showTabDropdown(chevronEl, wsId) {
|
||||
forkWorkstream(wsId);
|
||||
},
|
||||
},
|
||||
{
|
||||
label: "Export conversation",
|
||||
action: function () {
|
||||
exportWorkstreamDownload(wsId);
|
||||
},
|
||||
},
|
||||
{
|
||||
label: "Close",
|
||||
key: "Ctrl+W",
|
||||
|
||||
Reference in New Issue
Block a user