mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-28 06:44:51 -06:00
Patrick Buckley
97fbfb9f8e
* feat: workstream attachments (images + text documents)
Adds end-to-end support for attaching images (png/jpeg/gif/webp) and
plain-text documents (markdown, source, JSON, etc.) to a workstream's
next user turn via the web UI.
Storage: new workstream_attachments table (migration 037) with a
three-state lifecycle — pending → reserved → consumed — scoped by
(ws_id, user_id) and linked to conversations.id on consume. Rewind/
truncation cascades attachment rows; delete_workstream does too.
Session: ChatSession.send(attachments, send_id) builds multipart user
content (text + image_url + document parts) and persists text-only to
conversations with attachments joined on load via message_id. Queue
path carries ordered attachment_ids plus a reservation token so
queued multimodal turns can't lose files to overlapping sends.
Providers: internal document content parts translate at the API
boundary — Anthropic emits native document blocks (text/plain
coerced, original MIME folded into title); OpenAI Chat Completions
and the Google OpenAI-compat endpoint inline them as escaped
<document> text blocks (XML-attr escape + </document> neutralization);
Responses API emits input_text with the same wrapper.
Server: POST/GET/DELETE /v1/api/workstreams/{ws_id}/attachments with
multipart upload (magic-byte image sniffing, UTF-8 enforcement for
text, per-kind size caps, Content-Length pre-check, per-(ws,user)
pending cap + TOCTOU lock). /v1/api/send reserves before dispatch
using a full-UUID token, threads it into session.send / queue_message,
releases on worker-thread failure, and reports attached/dropped ids
so the UI can reflect partial reservations. GET /content sets
X-Content-Type-Options, CSP sandbox, inline Content-Disposition, and
forces text/plain for text kinds. Ownership failures mask as 404.
UI: paperclip button, hidden file input with accept allowlist, chip
strip above textarea, drag/drop + paste-image handlers. Chips
rehydrate on ws switch and on queued-message dequeue; send clears
only attached ids and shows a toast when some dropped. Historical
user messages render filename pills via a _attachments_meta sibling
populated on both live-send and reconstruct paths.
530 tests covering CRUD, reservation lifecycle, races (TOCTOU cap,
reserve-then-dispatch overlap), provider translation, XSS headers,
cascade delete, history round-trip, and service-scoped actor flow.
* fix(attachments): address PR review feedback
- get_attachment_content now scopes the row by user_id too, so an
unowned workstream can't be a vector for cross-user blob fetches
via attachment_id guessing (Copilot, server.py:2676)
- send_message rejects attachment_ids lists longer than the pending
cap with 400 — prevents hostile clients from blowing up the
storage IN (...) clause (Copilot, server.py:1515)
- _attachment_upload_locks switched to a bounded LRU OrderedDict;
evicts the oldest unlocked entries past the soft cap so the map
can't grow unboundedly on long-running nodes (Copilot, server.py:2417)
- Pane.dragleave handler uses relatedTarget instead of target so the
drop-zone styling clears correctly when the cursor moves through
child elements; dragend listener added as a fallback for cancelled
drags (Copilot, app.js:297)
- uploadAttachment always cleans up the placeholder chip on failure,
including auth errors — no more stuck "uploading..." chips after
re-auth (Copilot, app.js:427)
- New _swapPlaceholderChip / _removeAttachmentChip helpers preserve
user-selection order through the placeholder→real-id swap; the
pendingAttachments Map is rebuilt in place rather than naïvely
delete+set, which would have moved the entry to iteration end
(Copilot, app.js:420)
- Drop unused `var self = this;` in removeAttachment (github-code-quality)
- Two regression tests: cross-user fetch on an unowned workstream,
and oversized attachment_ids list rejection
* fix(attachments): switch upload-lock to threading.Lock to avoid 3.12 CI hang
The per-(ws, user) upload lock was a module-cached asyncio.Lock.
Starlette's TestClient runs each request on a fresh anyio task /
event loop, so the cached lock's internal _waiters bind to the first
loop that acquired it. When a later request runs in a different
loop, await lock.acquire() blocks on a Future from a closed loop —
silent deadlock.
This surfaced as test (3.12) hanging indefinitely in CI on one push
while the same suite passed on 3.11/3.13 and on the next push. Same
root cause is reproducible against any Starlette TestClient harness
on 3.10+; 3.12 just happens to surface it more often given changes
in how anyio + asyncio.Future interact across loop teardown.
Switched to threading.Lock — loop-agnostic, and the critical section
is one COUNT + one INSERT, short enough that briefly blocking the
event loop is fine. Updated the LRU-eviction probe accordingly
(threading.Lock has no public .locked(), so use a non-blocking
acquire+release as the "is it free?" probe).
TOCTOU pending-cap test still passes; full attachment suite passes
on both 3.12 and 3.13.
320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""Tests for the shared message reconstruction logic."""
|
|
|
|
import itertools
|
|
import json
|
|
|
|
from turnstone.core.storage._utils import reconstruct_messages
|
|
|
|
_row_ids = itertools.count(1)
|
|
|
|
|
|
def _row(
|
|
role,
|
|
content=None,
|
|
tool_name=None,
|
|
tc_id=None,
|
|
pdata=None,
|
|
tool_calls=None,
|
|
):
|
|
"""Build a 7-element conversation row tuple (id, role, ...)."""
|
|
return (next(_row_ids), role, content, tool_name, tc_id, pdata, tool_calls)
|
|
|
|
|
|
class TestAssistantWithToolCalls:
|
|
"""Assistant messages with tool_calls JSON are self-contained."""
|
|
|
|
def test_assistant_with_tool_calls_and_content(self):
|
|
tc = json.dumps(
|
|
[
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "read_file", "arguments": '{"path":"/tmp/x"}'},
|
|
}
|
|
]
|
|
)
|
|
rows = [
|
|
_row("assistant", "Let me check that.", tool_calls=tc),
|
|
_row("tool", "file contents", tc_id="call_1"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 2
|
|
assert msgs[0]["role"] == "assistant"
|
|
assert msgs[0]["content"] == "Let me check that."
|
|
assert len(msgs[0]["tool_calls"]) == 1
|
|
assert msgs[0]["tool_calls"][0]["function"]["name"] == "read_file"
|
|
assert msgs[1]["role"] == "tool"
|
|
assert msgs[1]["tool_call_id"] == "call_1"
|
|
|
|
def test_assistant_with_multiple_tool_calls(self):
|
|
tc = json.dumps(
|
|
[
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
|
|
},
|
|
{
|
|
"id": "call_2",
|
|
"type": "function",
|
|
"function": {"name": "bash", "arguments": '{"command":"pwd"}'},
|
|
},
|
|
]
|
|
)
|
|
rows = [
|
|
_row("assistant", tool_calls=tc),
|
|
_row("tool", "files", tc_id="call_1"),
|
|
_row("tool", "/home", tc_id="call_2"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 3
|
|
assert len(msgs[0]["tool_calls"]) == 2
|
|
assert msgs[1]["role"] == "tool"
|
|
assert msgs[2]["role"] == "tool"
|
|
|
|
def test_assistant_without_tool_calls(self):
|
|
rows = [_row("assistant", "Hello there.")]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["role"] == "assistant"
|
|
assert msgs[0]["content"] == "Hello there."
|
|
assert "tool_calls" not in msgs[0]
|
|
|
|
|
|
class TestMultipleTurns:
|
|
"""Multiple assistant turns with tool calls stay separate."""
|
|
|
|
def test_two_tool_call_turns(self):
|
|
tc1 = json.dumps(
|
|
[
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
|
|
}
|
|
]
|
|
)
|
|
tc2 = json.dumps(
|
|
[
|
|
{
|
|
"id": "call_2",
|
|
"type": "function",
|
|
"function": {"name": "bash", "arguments": '{"command":"cat file1"}'},
|
|
}
|
|
]
|
|
)
|
|
rows = [
|
|
_row("assistant", "I'll run two commands.", tool_calls=tc1),
|
|
_row("tool", "file1\nfile2", tc_id="call_1"),
|
|
_row("assistant", "Now reading.", tool_calls=tc2),
|
|
_row("tool", "contents", tc_id="call_2"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 4
|
|
assert msgs[0]["content"] == "I'll run two commands."
|
|
assert len(msgs[0]["tool_calls"]) == 1
|
|
assert msgs[2]["content"] == "Now reading."
|
|
assert len(msgs[2]["tool_calls"]) == 1
|
|
|
|
def test_denied_tool_calls_with_commentary(self):
|
|
"""Two denied tool batches with assistant commentary in between."""
|
|
tc1 = json.dumps(
|
|
[
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "bash", "arguments": '{"command":"find /"}'},
|
|
}
|
|
]
|
|
)
|
|
tc2 = json.dumps(
|
|
[
|
|
{
|
|
"id": "call_2",
|
|
"type": "function",
|
|
"function": {"name": "bash", "arguments": '{"command":"curl ..."}'},
|
|
}
|
|
]
|
|
)
|
|
rows = [
|
|
_row("assistant", tool_calls=tc1),
|
|
_row("tool", "Denied by user", tc_id="call_1"),
|
|
_row("assistant", "Interesting! Let me try something else."),
|
|
_row("assistant", tool_calls=tc2),
|
|
_row("tool", "Denied by user", tc_id="call_2"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 5
|
|
assert msgs[0]["role"] == "assistant"
|
|
assert msgs[0]["tool_calls"][0]["function"]["name"] == "bash"
|
|
assert msgs[1]["role"] == "tool"
|
|
assert msgs[2]["role"] == "assistant"
|
|
assert msgs[2]["content"] == "Interesting! Let me try something else."
|
|
assert "tool_calls" not in msgs[2]
|
|
assert msgs[3]["role"] == "assistant"
|
|
assert msgs[3]["tool_calls"][0]["function"]["name"] == "bash"
|
|
assert msgs[4]["role"] == "tool"
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Edge cases in message reconstruction."""
|
|
|
|
def test_incomplete_turn_repair(self):
|
|
"""Trailing tool_calls without enough tool_results are stripped."""
|
|
tc = json.dumps(
|
|
[
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
|
|
},
|
|
{
|
|
"id": "call_2",
|
|
"type": "function",
|
|
"function": {"name": "bash", "arguments": '{"command":"cat x"}'},
|
|
},
|
|
]
|
|
)
|
|
rows = [
|
|
_row("user", "hello"),
|
|
_row("assistant", "Let me check.", tool_calls=tc),
|
|
# Only 1 tool result for 2 tool_calls
|
|
_row("tool", "file1", tc_id="call_1"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["role"] == "user"
|
|
|
|
def test_empty_rows(self):
|
|
msgs = reconstruct_messages([], "ws1")
|
|
assert msgs == []
|
|
|
|
def test_provider_data_preserved(self):
|
|
pdata = json.dumps([{"type": "text", "text": "hello"}])
|
|
rows = [_row("assistant", "hello", pdata=pdata)]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert msgs[0]["_provider_content"] == [{"type": "text", "text": "hello"}]
|
|
|
|
def test_user_message(self):
|
|
rows = [_row("user", "hello world")]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 1
|
|
assert msgs[0] == {"role": "user", "content": "hello world"}
|
|
|
|
def test_none_content_becomes_empty_string(self):
|
|
rows = [_row("user", None)]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert msgs[0]["content"] == ""
|
|
|
|
def test_tool_without_tc_id_uses_empty_string(self):
|
|
rows = [
|
|
_row(
|
|
"assistant",
|
|
tool_calls=json.dumps(
|
|
[{"id": "c1", "type": "function", "function": {"name": "x", "arguments": ""}}]
|
|
),
|
|
),
|
|
_row("tool", "output", tc_id=None),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert msgs[1]["tool_call_id"] == ""
|
|
|
|
def test_unknown_role_ignored(self):
|
|
rows = [
|
|
_row("user", "hi"),
|
|
_row("system", "you are helpful"),
|
|
_row("assistant", "hello"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 2
|
|
assert msgs[0]["role"] == "user"
|
|
assert msgs[1]["role"] == "assistant"
|
|
|
|
|
|
class TestMidConversationOrphanRepair:
|
|
"""Mid-conversation orphaned tool_calls get synthetic tool results."""
|
|
|
|
def test_all_orphaned_mid_conversation(self):
|
|
"""Assistant has 2 tool_calls, no tool results, then user message."""
|
|
tc = json.dumps(
|
|
[
|
|
{"id": "c1", "function": {"name": "bash", "arguments": "{}"}},
|
|
{"id": "c2", "function": {"name": "write_file", "arguments": "{}"}},
|
|
]
|
|
)
|
|
rows = [
|
|
_row("user", "do stuff"),
|
|
_row("assistant", "Running...", tool_calls=tc),
|
|
_row("user", "never mind"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
# Should have: user, assistant, tool(c1), tool(c2), user
|
|
assert len(msgs) == 5
|
|
assert msgs[2]["role"] == "tool"
|
|
assert msgs[2]["tool_call_id"] == "c1"
|
|
assert msgs[2]["is_error"] is True
|
|
assert msgs[3]["role"] == "tool"
|
|
assert msgs[3]["tool_call_id"] == "c2"
|
|
assert msgs[4]["role"] == "user"
|
|
|
|
def test_partial_results_mid_conversation(self):
|
|
"""2 tool_calls, 1 result present, 1 missing — synthesize only the missing one."""
|
|
tc = json.dumps(
|
|
[
|
|
{"id": "c1", "function": {"name": "bash", "arguments": "{}"}},
|
|
{"id": "c2", "function": {"name": "write_file", "arguments": "{}"}},
|
|
]
|
|
)
|
|
rows = [
|
|
_row("user", "do stuff"),
|
|
_row("assistant", "", tool_calls=tc),
|
|
_row("tool", "file1.txt", tool_name="bash", tc_id="c1"),
|
|
_row("user", "skip the write"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
# Should have: user, assistant, tool(c1 real), tool(c2 synthetic), user
|
|
assert len(msgs) == 5
|
|
assert msgs[2]["role"] == "tool"
|
|
assert msgs[2]["tool_call_id"] == "c1"
|
|
assert msgs[2]["content"] == "file1.txt"
|
|
assert msgs[2].get("is_error") is not True
|
|
assert msgs[3]["role"] == "tool"
|
|
assert msgs[3]["tool_call_id"] == "c2"
|
|
assert msgs[3]["is_error"] is True
|
|
assert msgs[4]["role"] == "user"
|
|
|
|
def test_complete_results_no_synthesis(self):
|
|
"""All tool_calls have results — no synthesis needed."""
|
|
tc = json.dumps(
|
|
[
|
|
{"id": "c1", "function": {"name": "bash", "arguments": "{}"}},
|
|
]
|
|
)
|
|
rows = [
|
|
_row("user", "do it"),
|
|
_row("assistant", "", tool_calls=tc),
|
|
_row("tool", "done", tool_name="bash", tc_id="c1"),
|
|
_row("user", "thanks"),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
assert len(msgs) == 4
|
|
tool_msgs = [m for m in msgs if m["role"] == "tool"]
|
|
assert len(tool_msgs) == 1
|
|
assert tool_msgs[0].get("is_error") is not True
|
|
|
|
def test_trailing_orphan_stripped_not_synthesized(self):
|
|
"""Trailing orphan is handled by the existing strip repair, not synthesis."""
|
|
tc = json.dumps(
|
|
[
|
|
{"id": "c1", "function": {"name": "bash", "arguments": "{}"}},
|
|
]
|
|
)
|
|
rows = [
|
|
_row("user", "do it"),
|
|
_row("assistant", "Running...", tool_calls=tc),
|
|
]
|
|
msgs = reconstruct_messages(rows, "ws1")
|
|
# Trailing strip removes the assistant message entirely
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["role"] == "user"
|