mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
480a1426b3
* fix(session): fail-closed history-commit handoff (#981) The deleted-workstream discovery is now a terminal, ws_id-keyed latch: keyed conversation commits refuse admission once the durable parent is gone (convergence finalizers and force-abandon are exempt), history handoff refuses to mint a proof token so /history fails closed with a 503 instead of silently wiping the pane, and the SSE stream carries a workstream_gone resync reason. Discarded commits leave a forensic log of commit keys and roles, never content. Conversation rows gain a commit_key (migration 071): keyed saves are idempotent under retry, validated against the full commit identity, and refused when they would cross a workstream deletion. The prune orphan category now requires a NULL alias plus a two-hour updated grace, with cutoffs computed at discovery time and carried into both dialects' rechecks. The mid-turn interjection queue is owner-partitioned with no per-site mode flags: pops take the acting principal's and unowned rows, other participants' rows are structurally retained, and enforcement lives at queue admission plus the shared before_spawn gates. The retraction ledger is bounded by open pop windows: pops open a window atomically with the queue delete, restores close their ids atomically with the ledger consume, every other exit closes through one helper, and misses for unheld ids record nothing. The workstream-gone latch refuses unattended wakes at all three gates (watcher spawn, claim, delivery pre-pop), and the retry dispatcher regained its pre-envelope cancel/error convergence net. Persistence-state reporting derives through the session bound to each UI instead of a registry lookup by id that failed open to healthy during tombstone retention. The dashboard roster no longer re-inserts ghost entries from trailing activity events, the history tool-outcome scan tolerates interleaved non-turn rows, and the shared handoff-deadline handle owns its own retirement. Single-sourced across call sites: keyed-commit row values, attachment save wrappers, tail-truncation and conflict-resolution bodies for both storage dialects; worker-slot lifecycle field sets; the direct-commit admission frame; queued-row layout accessors; the string-aware comment stripper shared by every JS harness suite. Refs #981 #964 * fix(session): sweep handoff fixes to their sibling surfaces The interactive replay loop treated a system row as a tool-batch boundary, so every tool result after an interleaved row vanished from that pane while the coordinator rendered the same history correctly. Only a conversational turn ends the batch window now, matching the shared outcome index. Accepted user turns clear the composer's attachment chips on the same viewer policy that settles optimistic bubbles rather than on having matched a local bubble, so a workstream created with an upload no longer keeps a chip for an attachment the create dispatch already consumed. The coordinator's raced-Stop arm emits the stream-end hook it inherits alongside the idle state, leaving no unfinalized bubble or unflushed tool output. Ending a session surfaces a failure toast when the request never lands or answers with a non-JSON body. The per-second persistence reconcile now probes each session without blocking: a workstream whose generation and handoff locks are held is skipped until the next pass instead of contending the locks every commit needs. The one-shot repair that gates workstream creation at capacity keeps a definite probe — it has no next pass, and the sessions likeliest to be contended are the ones whose unresolved journals emptied its candidate list. Single-sourced: the attachment lane builds its conversation row through the shared commit-identity builder; the ordinary worker exit releases its slot through the lifecycle owner; both operator surfaces snapshot their counters through one non-consuming helper; the replay preamble loses its per-kind wrappers and its config hook; the browser harness suites share one brace walker; and each in-flight history attempt is one record carrying both its abort controller and its deadline. Refs #981 #964
483 lines
14 KiB
Python
483 lines
14 KiB
Python
"""Tests for turnstone.sdk.events — SSE event deserialization."""
|
|
|
|
from turnstone.sdk.events import (
|
|
ApprovalResolvedEvent,
|
|
ApproveRequestEvent,
|
|
BusyErrorEvent,
|
|
ClearUiEvent,
|
|
ClusterEvent,
|
|
ClusterStateEvent,
|
|
ClusterWsClosedEvent,
|
|
ClusterWsCreatedEvent,
|
|
ClusterWsRenameEvent,
|
|
ConnectedEvent,
|
|
ContentEvent,
|
|
ErrorEvent,
|
|
HistoryEvent,
|
|
HistoryResyncEvent,
|
|
InfoEvent,
|
|
NodeJoinedEvent,
|
|
NodeLostEvent,
|
|
OutputWarningEvent,
|
|
ReasoningEvent,
|
|
ServerEvent,
|
|
StatusEvent,
|
|
StreamEndEvent,
|
|
ThinkingStartEvent,
|
|
ThinkingStopEvent,
|
|
ToolInfoEvent,
|
|
ToolOutputChunkEvent,
|
|
ToolResultEvent,
|
|
UserTurnEvent,
|
|
WsActivityEvent,
|
|
WsClosedEvent,
|
|
WsRenameEvent,
|
|
WsStateEvent,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-workstream events
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_connected_event():
|
|
e = ServerEvent.from_dict(
|
|
{"type": "connected", "model": "gpt-5", "model_alias": "fast", "skip_permissions": True}
|
|
)
|
|
assert isinstance(e, ConnectedEvent)
|
|
assert e.model == "gpt-5"
|
|
assert e.model_alias == "fast"
|
|
assert e.skip_permissions is True
|
|
|
|
|
|
def test_history_event():
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
e = ServerEvent.from_dict({"type": "history", "messages": msgs})
|
|
assert isinstance(e, HistoryEvent)
|
|
assert e.messages == msgs
|
|
|
|
|
|
def test_history_resync_event_preserves_repair_reason():
|
|
e = ServerEvent.from_dict(
|
|
{
|
|
"type": "history_resync",
|
|
"ws_id": "ws1",
|
|
"reason": "handoff_mismatch",
|
|
}
|
|
)
|
|
assert isinstance(e, HistoryResyncEvent)
|
|
assert e.ws_id == "ws1"
|
|
assert e.reason == "handoff_mismatch"
|
|
|
|
|
|
def test_user_turn_event_preserves_correlation_and_attribution():
|
|
e = ServerEvent.from_dict(
|
|
{
|
|
"type": "user_turn",
|
|
"ws_id": "ws1",
|
|
"content": "hello",
|
|
"attachments": [
|
|
{
|
|
"attachment_id": "a1",
|
|
"kind": "text",
|
|
"filename": "note.txt",
|
|
"mime_type": "text/plain",
|
|
}
|
|
],
|
|
"sender": "user-1",
|
|
"client_send_ids": ["browser-send"],
|
|
"_event_id": 17,
|
|
}
|
|
)
|
|
assert isinstance(e, UserTurnEvent)
|
|
assert e.client_send_ids == ["browser-send"]
|
|
assert e.sender == "user-1"
|
|
assert e.attachments[0]["attachment_id"] == "a1"
|
|
assert e._event_id == 17
|
|
|
|
|
|
def test_thinking_start_stop():
|
|
e1 = ServerEvent.from_dict({"type": "thinking_start"})
|
|
e2 = ServerEvent.from_dict({"type": "thinking_stop"})
|
|
assert isinstance(e1, ThinkingStartEvent)
|
|
assert isinstance(e2, ThinkingStopEvent)
|
|
|
|
|
|
def test_content_event():
|
|
e = ServerEvent.from_dict({"type": "content", "text": "hello"})
|
|
assert isinstance(e, ContentEvent)
|
|
assert e.text == "hello"
|
|
|
|
|
|
def test_reasoning_event():
|
|
e = ServerEvent.from_dict({"type": "reasoning", "text": "step 1"})
|
|
assert isinstance(e, ReasoningEvent)
|
|
assert e.text == "step 1"
|
|
|
|
|
|
def test_stream_end_event():
|
|
e = ServerEvent.from_dict({"type": "stream_end"})
|
|
assert isinstance(e, StreamEndEvent)
|
|
|
|
|
|
def test_tool_info_event():
|
|
items = [{"name": "search", "call_id": "c1"}]
|
|
e = ServerEvent.from_dict({"type": "tool_info", "items": items})
|
|
assert isinstance(e, ToolInfoEvent)
|
|
assert e.items == items
|
|
|
|
|
|
def test_approve_request_event():
|
|
items = [{"name": "bash", "call_id": "c2", "arguments": "ls"}]
|
|
e = ServerEvent.from_dict({"type": "approve_request", "items": items})
|
|
assert isinstance(e, ApproveRequestEvent)
|
|
assert len(e.items) == 1
|
|
|
|
|
|
def test_approval_resolved_event():
|
|
e = ServerEvent.from_dict(
|
|
{"type": "approval_resolved", "approved": False, "feedback": "Approval timed out"}
|
|
)
|
|
assert isinstance(e, ApprovalResolvedEvent)
|
|
assert e.approved is False
|
|
assert e.feedback == "Approval timed out"
|
|
|
|
|
|
def test_tool_result_event():
|
|
e = ServerEvent.from_dict(
|
|
{"type": "tool_result", "call_id": "c1", "name": "search", "output": "found it"}
|
|
)
|
|
assert isinstance(e, ToolResultEvent)
|
|
assert e.call_id == "c1"
|
|
assert e.name == "search"
|
|
assert e.output == "found it"
|
|
|
|
|
|
def test_accepted_tool_result_event_carries_final_projection_metadata():
|
|
preview = {"kind": "html", "attachment_id": "preview-1"}
|
|
e = ServerEvent.from_dict(
|
|
{
|
|
"type": "tool_result",
|
|
"call_id": "c-final",
|
|
"name": "open_preview",
|
|
"output": "guarded\nscalar",
|
|
"is_error": True,
|
|
"preview": preview,
|
|
"accepted": True,
|
|
"effect_status": "unknown",
|
|
"_event_id": 42,
|
|
}
|
|
)
|
|
|
|
assert isinstance(e, ToolResultEvent)
|
|
assert e.output == "guarded\nscalar"
|
|
assert e.is_error is True
|
|
assert e.preview == preview
|
|
assert e.accepted is True
|
|
assert e.effect_status == "unknown"
|
|
assert e._event_id == 42
|
|
|
|
|
|
def test_tool_output_chunk_event():
|
|
e = ServerEvent.from_dict({"type": "tool_output_chunk", "call_id": "c1", "chunk": "line1\n"})
|
|
assert isinstance(e, ToolOutputChunkEvent)
|
|
assert e.chunk == "line1\n"
|
|
|
|
|
|
def test_output_warning_event_llm_tier():
|
|
"""LLM-tier finding carries confidence + reasoning + judge_model so SDK
|
|
consumers see the same attribution the UI chip renders."""
|
|
e = ServerEvent.from_dict(
|
|
{
|
|
"type": "output_warning",
|
|
"call_id": "c1",
|
|
"func_name": "web_fetch",
|
|
"risk_level": "medium",
|
|
"flags": ["camouflaged_injection"],
|
|
"redacted": False,
|
|
"tier": "llm",
|
|
"judge_risk": "none",
|
|
"confidence": 0.82,
|
|
"reasoning": "Authority-framed directive embedded in the doc.",
|
|
"judge_model": "gpt-5-mini",
|
|
}
|
|
)
|
|
assert isinstance(e, OutputWarningEvent)
|
|
assert e.tier == "llm"
|
|
assert e.judge_risk == "none" # the judge's OWN verdict (may differ from risk_level)
|
|
assert e.confidence == 0.82
|
|
assert e.flags == ["camouflaged_injection"]
|
|
assert e.reasoning == "Authority-framed directive embedded in the doc."
|
|
assert e.judge_model == "gpt-5-mini"
|
|
|
|
|
|
def test_output_warning_event_heuristic_defaults():
|
|
"""A regex-only finding defaults tier=heuristic with no confidence."""
|
|
e = ServerEvent.from_dict(
|
|
{"type": "output_warning", "call_id": "c1", "risk_level": "high", "redacted": True}
|
|
)
|
|
assert isinstance(e, OutputWarningEvent)
|
|
assert e.tier == "heuristic"
|
|
assert e.confidence == 0.0
|
|
assert e.redacted is True
|
|
|
|
|
|
def test_output_warning_event_covers_every_merge_payload_key():
|
|
"""Drift guard: every key the server-side merge can emit must be a declared
|
|
OutputWarningEvent field, else from_dict silently drops it (the bug that let
|
|
`annotations` go stale). Builds the maximal payload and checks the field-set."""
|
|
import dataclasses
|
|
|
|
from turnstone.core.output_guard import merge_guard_display_payload
|
|
|
|
# Maximal payload — every optional field populated.
|
|
payload = merge_guard_display_payload(
|
|
heuristic_risk="high",
|
|
heuristic_flags=["credential_leak"],
|
|
heuristic_annotations=["API key detected."],
|
|
redacted=True,
|
|
llm_succeeded=True,
|
|
llm_risk="none",
|
|
llm_flags=["camouflaged_injection"],
|
|
llm_reasoning="Benign.",
|
|
llm_confidence=0.9,
|
|
llm_model="gpt-5-mini",
|
|
)
|
|
assert payload is not None
|
|
declared = {f.name for f in dataclasses.fields(OutputWarningEvent)}
|
|
missing = set(payload) - declared
|
|
assert not missing, f"OutputWarningEvent is missing merge-payload fields: {missing}"
|
|
|
|
|
|
def test_status_event():
|
|
e = ServerEvent.from_dict(
|
|
{
|
|
"type": "status",
|
|
"prompt_tokens": 100,
|
|
"completion_tokens": 50,
|
|
"total_tokens": 150,
|
|
"context_window": 128000,
|
|
"pct": 0.12,
|
|
"effort": "medium",
|
|
}
|
|
)
|
|
assert isinstance(e, StatusEvent)
|
|
assert e.prompt_tokens == 100
|
|
assert e.total_tokens == 150
|
|
assert e.pct == 0.12
|
|
assert e.effort == "medium"
|
|
|
|
|
|
def test_info_event():
|
|
e = ServerEvent.from_dict({"type": "info", "message": "[compacted]"})
|
|
assert isinstance(e, InfoEvent)
|
|
assert e.message == "[compacted]"
|
|
|
|
|
|
def test_error_event():
|
|
e = ServerEvent.from_dict({"type": "error", "message": "Something broke"})
|
|
assert isinstance(e, ErrorEvent)
|
|
assert e.message == "Something broke"
|
|
|
|
|
|
def test_busy_error_event():
|
|
e = ServerEvent.from_dict({"type": "busy_error", "message": "Already processing a request."})
|
|
assert isinstance(e, BusyErrorEvent)
|
|
assert "Already" in e.message
|
|
|
|
|
|
def test_clear_ui_event():
|
|
e = ServerEvent.from_dict({"type": "clear_ui"})
|
|
assert isinstance(e, ClearUiEvent)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Global events
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_ws_state_event():
|
|
e = ServerEvent.from_dict(
|
|
{
|
|
"type": "ws_state",
|
|
"ws_id": "ws1",
|
|
"state": "thinking",
|
|
"tokens": 500,
|
|
"context_ratio": 0.3,
|
|
"activity": "Writing code",
|
|
"activity_state": "thinking",
|
|
"persistence_state": "retrying",
|
|
}
|
|
)
|
|
assert isinstance(e, WsStateEvent)
|
|
assert e.ws_id == "ws1"
|
|
assert e.state == "thinking"
|
|
assert e.tokens == 500
|
|
assert e.persistence_state == "retrying"
|
|
|
|
|
|
def test_ws_state_event_defaults_persistence_for_older_nodes():
|
|
e = ServerEvent.from_dict({"type": "ws_state", "ws_id": "ws1"})
|
|
assert isinstance(e, WsStateEvent)
|
|
assert e.persistence_state == "healthy"
|
|
|
|
|
|
def test_ws_activity_event():
|
|
e = ServerEvent.from_dict(
|
|
{"type": "ws_activity", "ws_id": "ws1", "activity": "reading", "activity_state": "tool"}
|
|
)
|
|
assert isinstance(e, WsActivityEvent)
|
|
assert e.activity == "reading"
|
|
|
|
|
|
def test_ws_rename_event():
|
|
e = ServerEvent.from_dict({"type": "ws_rename", "ws_id": "ws1", "name": "My Chat"})
|
|
assert isinstance(e, WsRenameEvent)
|
|
assert e.name == "My Chat"
|
|
|
|
|
|
def test_ws_closed_event():
|
|
e = ServerEvent.from_dict({"type": "ws_closed", "ws_id": "ws1", "name": "old"})
|
|
assert isinstance(e, WsClosedEvent)
|
|
assert e.name == "old"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cluster events
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_node_joined_event():
|
|
e = ClusterEvent.from_dict({"type": "node_joined", "node_id": "host1_abc"})
|
|
assert isinstance(e, NodeJoinedEvent)
|
|
assert e.node_id == "host1_abc"
|
|
|
|
|
|
def test_node_lost_event():
|
|
e = ClusterEvent.from_dict({"type": "node_lost", "node_id": "host2_def"})
|
|
assert isinstance(e, NodeLostEvent)
|
|
assert e.node_id == "host2_def"
|
|
|
|
|
|
def test_cluster_state_event():
|
|
e = ClusterEvent.from_dict(
|
|
{
|
|
"type": "cluster_state",
|
|
"ws_id": "ws1",
|
|
"node_id": "n1",
|
|
"state": "running",
|
|
"tokens": 1000,
|
|
"context_ratio": 0.5,
|
|
"activity": "executing tool",
|
|
"activity_state": "tool",
|
|
"persistence_state": "conflict",
|
|
}
|
|
)
|
|
assert isinstance(e, ClusterStateEvent)
|
|
assert e.node_id == "n1"
|
|
assert e.state == "running"
|
|
assert e.tokens == 1000
|
|
assert e.persistence_state == "conflict"
|
|
|
|
|
|
def test_cluster_ws_created_event():
|
|
e = ClusterEvent.from_dict(
|
|
{
|
|
"type": "ws_created",
|
|
"ws_id": "ws2",
|
|
"node_id": "n1",
|
|
"name": "New WS",
|
|
"persistence_state": "pending",
|
|
}
|
|
)
|
|
assert isinstance(e, ClusterWsCreatedEvent)
|
|
assert e.ws_id == "ws2"
|
|
assert e.name == "New WS"
|
|
assert e.persistence_state == "pending"
|
|
|
|
|
|
def test_cluster_ws_closed_event():
|
|
e = ClusterEvent.from_dict({"type": "ws_closed", "ws_id": "ws2"})
|
|
assert isinstance(e, ClusterWsClosedEvent)
|
|
assert e.ws_id == "ws2"
|
|
|
|
|
|
def test_cluster_ws_rename_event():
|
|
e = ClusterEvent.from_dict({"type": "ws_rename", "ws_id": "ws2", "name": "Renamed"})
|
|
assert isinstance(e, ClusterWsRenameEvent)
|
|
assert e.name == "Renamed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_unknown_server_event_falls_back():
|
|
e = ServerEvent.from_dict({"type": "future_event", "ws_id": "ws1"})
|
|
assert type(e) is ServerEvent
|
|
assert e.type == "future_event"
|
|
assert e.ws_id == "ws1"
|
|
|
|
|
|
def test_unknown_cluster_event_falls_back():
|
|
e = ClusterEvent.from_dict({"type": "future_cluster_event"})
|
|
assert type(e) is ClusterEvent
|
|
assert e.type == "future_cluster_event"
|
|
|
|
|
|
def test_extra_fields_ignored():
|
|
e = ServerEvent.from_dict({"type": "content", "text": "hi", "extra_field": 999})
|
|
assert isinstance(e, ContentEvent)
|
|
assert e.text == "hi"
|
|
|
|
|
|
def test_in_progress_snapshot_event_round_trip():
|
|
from turnstone.sdk.events import InProgressSnapshotEvent
|
|
|
|
payload = {
|
|
"type": "in_progress_snapshot",
|
|
"ws_id": "ws1",
|
|
"content": "Partial content...",
|
|
"reasoning": "Partial reasoning...",
|
|
}
|
|
e = ServerEvent.from_dict(payload)
|
|
assert isinstance(e, InProgressSnapshotEvent)
|
|
assert e.ws_id == "ws1"
|
|
assert e.content == "Partial content..."
|
|
assert e.reasoning == "Partial reasoning..."
|
|
|
|
|
|
def test_in_progress_snapshot_event_strips_internal_seq():
|
|
"""``_seq`` is server-internal plumbing — even if a stray copy
|
|
leaks through, ``from_dict`` must drop it (not a declared field)."""
|
|
from turnstone.sdk.events import InProgressSnapshotEvent
|
|
|
|
e = ServerEvent.from_dict(
|
|
{
|
|
"type": "in_progress_snapshot",
|
|
"ws_id": "ws1",
|
|
"content": "x",
|
|
"reasoning": "",
|
|
"_seq": 42,
|
|
}
|
|
)
|
|
assert isinstance(e, InProgressSnapshotEvent)
|
|
assert not hasattr(e, "_seq")
|
|
|
|
|
|
def test_state_change_event_round_trip():
|
|
from turnstone.sdk.events import StateChangeEvent
|
|
|
|
e = ServerEvent.from_dict({"type": "state_change", "ws_id": "ws1", "state": "thinking"})
|
|
assert isinstance(e, StateChangeEvent)
|
|
assert e.state == "thinking"
|
|
assert e.ws_id == "ws1"
|
|
|
|
|
|
def test_missing_type_defaults_to_base():
|
|
e = ServerEvent.from_dict({"ws_id": "ws1"})
|
|
assert type(e) is ServerEvent
|
|
assert e.ws_id == "ws1"
|
|
assert e.type == ""
|