ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-12 23:12:23 -06:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
turnstone/tests/test_sessions.py
T
Patrick Buckley 480a1426b3 Fail-closed history-commit handoff (#1005) 
* 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
2026-08-11 04:18:36 -07:00

1538 lines
60 KiB
Python
Raw Permalink Blame History

"""Tests for workstream persistence and resume functionality."""
from unittest.mock import MagicMock, patch
import sqlalchemy as sa
from tests._oidc_test_helpers import keyed_app_state
from turnstone.core import model_registry as mr_module
from turnstone.core.memory import (
delete_workstream,
list_workstreams_with_history,
load_messages,
load_workstream_config,
prune_workstreams,
register_workstream,
resolve_workstream,
save_message,
save_workstream_config,
set_workstream_alias,
update_workstream_title,
)
from turnstone.core.model_turn import resolve_model_binding
from turnstone.core.session import ChatSession
from turnstone.core.storage import get_storage
from turnstone.core.trajectory import turn_to_dict
# ── Workstream registration ───────────────────────────────────────────
class TestRegisterWorkstream:
def test_register_creates_row(self, tmp_db):
register_workstream("abc123")
# Workstream exists in DB (resolve works) even without messages
assert resolve_workstream("abc123") == "abc123"
def test_register_with_title(self, tmp_db):
register_workstream("abc123", name="My Workstream")
save_message("abc123", "user", "hello")
rows = list_workstreams_with_history()
assert rows[0][2] is None # title column (name is separate)
def test_register_idempotent(self, tmp_db):
register_workstream("abc123")
update_workstream_title("abc123", "First")
register_workstream("abc123") # should be ignored
update_workstream_title("abc123", "First") # title is set via update
save_message("abc123", "user", "hello")
rows = list_workstreams_with_history()
assert len(rows) == 1
assert rows[0][2] == "First" # title preserved
def test_update_title(self, tmp_db):
register_workstream("abc123")
update_workstream_title("abc123", "New Title")
save_message("abc123", "user", "hello")
rows = list_workstreams_with_history()
assert rows[0][2] == "New Title"
# ── Workstream alias ──────────────────────────────────────────────────
class TestWorkstreamAlias:
def test_set_alias(self, tmp_db):
register_workstream("abc123")
assert set_workstream_alias("abc123", "my-session") is True
save_message("abc123", "user", "hello")
rows = list_workstreams_with_history()
assert rows[0][1] == "my-session" # alias
def test_alias_conflict(self, tmp_db):
register_workstream("abc123")
register_workstream("def456")
set_workstream_alias("abc123", "taken")
assert set_workstream_alias("def456", "taken") is False
def test_alias_same_workstream_ok(self, tmp_db):
register_workstream("abc123")
set_workstream_alias("abc123", "mine")
assert set_workstream_alias("abc123", "mine") is True # no-op, same workstream
# ── Workstream resolution ─────────────────────────────────────────────
class TestResolveWorkstream:
def test_resolve_by_alias(self, tmp_db):
register_workstream("abc123")
set_workstream_alias("abc123", "my-alias")
assert resolve_workstream("my-alias") == "abc123"
def test_resolve_by_exact_id(self, tmp_db):
register_workstream("abc123def456")
assert resolve_workstream("abc123def456") == "abc123def456"
def test_resolve_by_prefix(self, tmp_db):
register_workstream("abc123def456")
assert resolve_workstream("abc123") == "abc123def456"
def test_resolve_prefix_ambiguous(self, tmp_db):
register_workstream("abc123aaaaaa")
register_workstream("abc123bbbbbb")
# Ambiguous prefix should return None
assert resolve_workstream("abc123") is None
def test_resolve_not_found(self, tmp_db):
assert resolve_workstream("nonexistent") is None
# ── List workstreams with history ──────────────────────────────────────
class TestListWorkstreamsWithHistory:
def test_empty(self, tmp_db):
assert list_workstreams_with_history() == []
def test_ordered_by_updated(self, tmp_db):
register_workstream("first")
save_message("first", "user", "hello")
# Force an older timestamp so ordering is deterministic
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
conn.execute(
sa.text("UPDATE workstreams SET updated = '2020-01-01' WHERE ws_id = 'first'")
)
conn.commit()
register_workstream("second")
save_message("second", "user", "hello")
# second is more recent
rows = list_workstreams_with_history()
assert rows[0][0] == "second"
assert rows[1][0] == "first"
def test_includes_message_count(self, tmp_db):
register_workstream("sess1")
save_message("sess1", "user", "hello")
save_message("sess1", "assistant", "hi")
rows = list_workstreams_with_history()
assert rows[0][6] == 2 # msg_count (after ws_id, alias, title, name, created, updated)
def test_respects_limit(self, tmp_db):
for i in range(5):
register_workstream(f"sess{i}")
save_message(f"sess{i}", "user", "hello")
rows = list_workstreams_with_history(limit=3)
assert len(rows) == 3
# ── Load messages ─────────────────────────────────────────────────────
class TestLoadMessages:
def test_simple_user_assistant(self, tmp_db):
save_message("s1", "user", "hello")
save_message("s1", "assistant", "hi there")
msgs = load_messages("s1")
assert len(msgs) == 2
assert msgs[0] == {"role": "user", "content": "hello"}
assert msgs[1] == {"role": "assistant", "content": "hi there"}
def test_tool_calls_with_ids(self, tmp_db):
import json
tc_json = json.dumps(
[
{
"id": "call_abc",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
}
]
)
save_message("s1", "user", "run ls")
save_message("s1", "assistant", "Let me check.", tool_calls=tc_json)
save_message("s1", "tool", "file1.txt\nfile2.txt", "bash", tool_call_id="call_abc")
msgs = load_messages("s1")
assert len(msgs) == 3 # user, assistant+tool_calls, tool
# Assistant should have content and tool_calls
assert msgs[1]["role"] == "assistant"
assert msgs[1]["content"] == "Let me check."
assert len(msgs[1]["tool_calls"]) == 1
assert msgs[1]["tool_calls"][0]["id"] == "call_abc"
assert msgs[1]["tool_calls"][0]["function"]["name"] == "bash"
# Tool result
assert msgs[2]["role"] == "tool"
assert msgs[2]["tool_call_id"] == "call_abc"
assert msgs[2]["content"] == "file1.txt\nfile2.txt"
def test_parallel_tool_calls(self, tmp_db):
import json
tc_json = json.dumps(
[
{
"id": "call_1",
"type": "function",
"function": {"name": "search", "arguments": '{"query":"a"}'},
},
{
"id": "call_2",
"type": "function",
"function": {"name": "search", "arguments": '{"query":"b"}'},
},
]
)
save_message("s1", "user", "search two things")
save_message("s1", "assistant", None, tool_calls=tc_json)
save_message("s1", "tool", "result a", "search", tool_call_id="call_1")
save_message("s1", "tool", "result b", "search", tool_call_id="call_2")
msgs = load_messages("s1")
assert len(msgs) == 4 # user, assistant+2 tool_calls, 2 tool results
assert len(msgs[1]["tool_calls"]) == 2
assert msgs[2]["tool_call_id"] == "call_1"
assert msgs[3]["tool_call_id"] == "call_2"
def test_empty_workstream(self, tmp_db):
assert load_messages("nonexistent") == []
# ── Delete workstream ─────────────────────────────────────────────────
class TestDeleteWorkstream:
def test_delete_removes_workstream_and_messages(self, tmp_db):
register_workstream("abc123")
save_message("abc123", "user", "hello")
save_message("abc123", "assistant", "hi")
assert delete_workstream("abc123") is True
assert list_workstreams_with_history() == []
assert load_messages("abc123") == []
def test_delete_nonexistent(self, tmp_db):
assert delete_workstream("nonexistent") is False
# ── save_message with tool_call_id ────────────────────────────────────
class TestSaveMessageToolCallId:
def test_tool_call_id_stored(self, tmp_db):
save_message("s1", "tool", "output", "bash", tool_call_id="call_xyz")
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
row = conn.execute(
sa.text("SELECT tool_call_id FROM conversations WHERE ws_id = 's1'")
).fetchone()
assert row[0] == "call_xyz"
def test_tool_call_id_none_by_default(self, tmp_db):
save_message("s1", "user", "hello")
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
row = conn.execute(
sa.text("SELECT tool_call_id FROM conversations WHERE ws_id = 's1'")
).fetchone()
assert row[0] is None
# ── Workstreams table creation ────────────────────────────────────────
class TestWorkstreamsTable:
def test_workstreams_table_exists(self, tmp_db):
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
rows = conn.execute(
sa.text("SELECT name FROM sqlite_master WHERE type='table' AND name='workstreams'")
).fetchall()
assert len(rows) == 1
def test_tool_call_id_column_exists(self, tmp_db):
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
# Should not raise
conn.execute(sa.text("SELECT tool_call_id FROM conversations LIMIT 0"))
# ── ChatSession.resume ────────────────────────────────────────────────
class TestResumeWorkstream:
def test_resume_loads_messages(self, tmp_db, mock_openai_client):
# Set up a workstream with messages in DB
register_workstream("old_ws_123")
save_message("old_ws_123", "user", "hello world")
save_message("old_ws_123", "assistant", "hi there")
# Create a new session and resume
session = ChatSession(
client=mock_openai_client,
model="test-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
original_id = session._ws_id
assert original_id != "old_ws_123"
result = session.resume("old_ws_123")
assert result is True
assert session._ws_id == "old_ws_123"
assert len(session.messages) == 2
assert turn_to_dict(session.messages[0])["content"] == "hello world"
assert session._title_generated is True
def test_resume_nonexistent_returns_false(self, tmp_db, mock_openai_client):
session = ChatSession(
client=mock_openai_client,
model="test-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
assert session.resume("nonexistent") is False
def test_workstream_not_registered_until_message(self, tmp_db, mock_openai_client):
session = ChatSession(
client=mock_openai_client,
model="test-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
# Workstream is not auto-registered on init — only on /new or server creation
assert resolve_workstream(session._ws_id) is None
assert not any(r[0] == session._ws_id for r in list_workstreams_with_history())
# ── save_message updates workstreams.updated ──────────────────────────
class TestSaveMessageUpdatesWorkstream:
def test_updated_timestamp_bumped(self, tmp_db):
register_workstream("s1")
save_message("s1", "user", "first")
import time
time.sleep(0.01) # ensure different timestamp
save_message("s1", "user", "hello")
rows = list_workstreams_with_history()
new_updated = rows[0][4]
# updated should be same or later (sqlite datetime resolution is seconds,
# so they may be equal in fast tests — just verify no error)
assert new_updated is not None
# ── Interrupted workstream repair ─────────────────────────────────────
class TestInterruptedWorkstreamRepair:
"""load_messages() should strip trailing incomplete tool call turns."""
def test_complete_tool_turn_preserved(self, tmp_db):
"""2 tool_calls + 2 tool results = complete, no stripping."""
import json
tc_json = json.dumps(
[
{
"id": "call_1",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
},
{
"id": "call_2",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"pwd"}'},
},
]
)
save_message("s1", "user", "hello")
save_message("s1", "assistant", None, tool_calls=tc_json)
save_message("s1", "tool", "file.txt", tool_call_id="call_1")
save_message("s1", "tool", "/home", tool_call_id="call_2")
msgs = load_messages("s1")
assert len(msgs) == 4 # user + assistant(2 calls) + 2 tool results
def test_partial_tool_results_stripped(self, tmp_db):
"""2 tool_calls + 1 tool result = incomplete, strip the turn."""
import json
tc_json = json.dumps(
[
{
"id": "call_1",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
},
{
"id": "call_2",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"pwd"}'},
},
]
)
save_message("s1", "user", "hello")
save_message("s1", "assistant", None, tool_calls=tc_json)
save_message("s1", "tool", "file.txt", tool_call_id="call_1")
msgs = load_messages("s1")
assert len(msgs) == 1 # only user message remains
assert msgs[0]["role"] == "user"
def test_zero_tool_results_stripped(self, tmp_db):
"""Assistant with tool_calls + 0 results = incomplete, strip the turn."""
import json
tc_json = json.dumps(
[
{
"id": "call_1",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
},
{
"id": "call_2",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"pwd"}'},
},
]
)
save_message("s1", "user", "hello")
save_message("s1", "assistant", "Let me check", tool_calls=tc_json)
msgs = load_messages("s1")
assert len(msgs) == 1
assert msgs[0]["role"] == "user"
def test_complete_turn_before_incomplete_preserved(self, tmp_db):
"""Complete turn followed by incomplete turn: keep complete, strip incomplete."""
import json
tc_json = json.dumps(
[
{
"id": "call_1",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
},
]
)
save_message("s1", "user", "first")
save_message("s1", "assistant", "response")
save_message("s1", "user", "second")
save_message("s1", "assistant", None, tool_calls=tc_json)
msgs = load_messages("s1")
assert len(msgs) == 3 # user + assistant + user (incomplete turn stripped)
assert msgs[0]["role"] == "user"
assert msgs[1]["role"] == "assistant"
assert msgs[2]["role"] == "user"
def test_repair_false_preserves_partial_trailing_turn(self, tmp_db):
"""``repair=False`` is the display-read contract for ``/history``.
The default repair pass strips the trailing
``assistant(tool_calls)`` when not all tool results are persisted
— correct for ``session.resume`` (LLM context), wrong for the
REST display read. A user refreshing the coordinator page mid-
tool-execution would otherwise lose the entire trailing turn
from the UI. ``repair=False`` returns the raw persisted state.
"""
import json
tc_json = json.dumps(
[
{
"id": "call_1",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
},
{
"id": "call_2",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"pwd"}'},
},
]
)
save_message("s1", "user", "hello")
save_message("s1", "assistant", "Checking", tool_calls=tc_json)
save_message("s1", "tool", "file.txt", tool_call_id="call_1")
# No call_2 result persisted — mid-execution refresh.
msgs = get_storage().load_messages("s1", repair=False)
# All three rows survive — the trailing partial turn is what the
# operator was actually watching live.
assert [m["role"] for m in msgs] == ["user", "assistant", "tool"]
assert msgs[1].get("tool_calls") and len(msgs[1]["tool_calls"]) == 2
assert msgs[2]["tool_call_id"] == "call_1"
def test_repair_false_does_not_synthesize_orphan_results(self, tmp_db):
"""``repair=False`` must NOT splice synthetic ``"Tool execution
was cancelled."`` rows for mid-conversation orphans either —
the operator never saw those rows, and showing them would
invent UI content that doesn't reflect persisted state.
"""
import json
tc_json = json.dumps(
[
{
"id": "call_1",
"type": "function",
"function": {"name": "bash", "arguments": '{"command":"ls"}'},
},
]
)
save_message("s1", "user", "first")
save_message("s1", "assistant", "Working", tool_calls=tc_json)
# Cancel landed before any tool result — next turn happens.
save_message("s1", "user", "second")
save_message("s1", "assistant", "ok")
msgs = get_storage().load_messages("s1", repair=False)
roles = [m["role"] for m in msgs]
# No synthetic tool row spliced after the orphaned tool_calls.
assert roles == ["user", "assistant", "user", "assistant"]
assert all(m["role"] != "tool" for m in msgs)
# ── Workstream config persistence ─────────────────────────────────────
class TestWorkstreamConfig:
def test_save_load_roundtrip(self, tmp_db):
config = {
"temperature": "0.3",
"reasoning_effort": "high",
"persona": "scribe",
"persona_prompt": "You are a scribe.",
"persona_tools": "[]",
"persona_mcp": "0",
"persona_memory": "0",
}
save_workstream_config("s1", config)
loaded = load_workstream_config("s1")
assert loaded == config
def test_update_existing_key(self, tmp_db):
save_workstream_config("s1", {"temperature": "0.3"})
save_workstream_config("s1", {"temperature": "0.7"})
loaded = load_workstream_config("s1")
assert loaded["temperature"] == "0.7"
def test_missing_workstream_returns_empty(self, tmp_db):
loaded = load_workstream_config("nonexistent")
assert loaded == {}
def test_delete_workstream_removes_config(self, tmp_db):
register_workstream("s1")
save_message("s1", "user", "hi")
save_workstream_config("s1", {"temperature": "0.5"})
delete_workstream("s1")
assert load_workstream_config("s1") == {}
def test_resume_restores_config(self, tmp_db):
"""ChatSession.resume() should restore persisted config."""
client = MagicMock()
client.models.list.return_value.data = [MagicMock(id="test-model")]
ui = MagicMock()
ui.on_info = MagicMock()
ui.on_error = MagicMock()
ui.on_state_change = MagicMock()
ui.on_rename = MagicMock()
# Create a workstream with specific config
register_workstream("orig")
save_message("orig", "user", "hello")
save_message("orig", "assistant", "hi there")
save_workstream_config(
"orig",
{
"temperature": "0.3",
"reasoning_effort": "high",
"max_tokens": "2048",
"instructions": "be concise",
"persona": "writer",
"persona_prompt": "You are a creative writing partner.",
"persona_tools": "[]",
"persona_mcp": "0",
"persona_memory": "1",
},
)
# Create a new session with different defaults, then resume
session = ChatSession(
client=client,
model="test",
ui=ui,
instructions=None,
temperature=0.7,
max_tokens=4096,
tool_timeout=30,
)
assert session.temperature == 0.7 # default
assert session._persona_name == "" # unstamped constructor default
result = session.resume("orig")
assert result is True
assert session.temperature == 0.3
assert session.reasoning_effort == "high"
assert session.max_tokens == 2048
assert session.instructions == "be concise"
# Non-fork resume adopts the target's persona stamp so a later
# _save_config can't clobber it with this session's own stamp.
assert session._persona_name == "writer"
assert session._persona_prompt == "You are a creative writing partner."
assert session._persona_tools == frozenset()
assert session._persona_mcp is False
assert session._persona_memory is True
def test_resume_keeps_defaults_when_alias_unresolvable(self, tmp_db):
"""When the saved alias is empty or no longer in the registry,
``resume()`` must NOT copy ``saved_model`` onto the constructor's
default provider. Pairing a removed model name with a default
provider that doesn't know about it produces a broken session
whose next API call fails — the exact regression Copilot flagged
on PR #465. The constructor already resolved a coherent default
(provider + model + capabilities); resume should leave it intact
and just log the unreachable saved values."""
client = MagicMock()
client.models.list.return_value.data = [MagicMock(id="test-model")]
ui = MagicMock()
ui.on_info = MagicMock()
ui.on_error = MagicMock()
ui.on_state_change = MagicMock()
ui.on_rename = MagicMock()
register_workstream("model_ws")
save_message("model_ws", "user", "hello")
save_message("model_ws", "assistant", "hi")
# Empty alias + an orphan model name — same shape resume sees
# when an operator removes an alias from the registry that the
# workstream was originally pinned to.
save_workstream_config("model_ws", {"model": "gpt-5", "model_alias": ""})
session = ChatSession(
client=client,
model="gpt-5-nano",
ui=ui,
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
assert session.model == "gpt-5-nano"
result = session.resume("model_ws")
assert result is True
# Constructor's coherent default is preserved — saved orphan
# model name is NOT copied over.
assert session.model == "gpt-5-nano"
def test_resume_restore_stamps_current_generation(self, tmp_db):
from turnstone.core.model_registry import ModelConfig, ModelRegistry
reg = ModelRegistry(
models={
"a": ModelConfig("a", "http://a/v1", "k", "m-a"),
"b": ModelConfig("b", "http://b/v1", "k", "m-b"),
},
default="a",
)
register_workstream("gen_ws")
save_message("gen_ws", "user", "hello")
save_workstream_config("gen_ws", {"model": "m-b", "model_alias": "b"})
binding = resolve_model_binding(reg, "a")
session = ChatSession(
client=binding.lane.client,
model=binding.lane.model,
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
registry=reg,
model_alias="a",
model_binding=binding,
)
reg.reload(
{
"a": ModelConfig("a", "http://a/v1", "k", "m-a"),
"b": ModelConfig("b", "http://b/v1", "k", "m-b"),
},
"a",
app_state=keyed_app_state(),
)
assert session.resume("gen_ws") is True
assert session.model == "m-b"
binding = session._model_binding
assert binding.lane.client is reg.get_client("b")
assert binding.lane.provider is reg.get_provider("b")
assert binding.config is reg.get_config("b")
assert binding.registry_generation == reg.generation
def test_resume_keeps_binding_when_alias_vanishes_mid_restore(self, tmp_db):
"""The has_alias/resolve straddle must not raise out of resume."""
from turnstone.core.model_registry import ModelConfig, ModelRegistry
reg = ModelRegistry(
models={"a": ModelConfig("a", "http://a/v1", "k", "m-a")},
default="a",
)
register_workstream("race_ws")
save_message("race_ws", "user", "hello")
save_workstream_config("race_ws", {"model": "m-a", "model_alias": "a"})
binding = resolve_model_binding(reg, "a")
session = ChatSession(
client=binding.lane.client,
model=binding.lane.model,
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
registry=reg,
model_alias="a",
model_binding=binding,
)
old_binding = session._model_binding
# has_alias passes, then the resolve finds the alias gone — the
# straddle a concurrent reload produces.
with patch.object(reg, "resolve_binding", side_effect=ValueError("Unknown model alias: a")):
assert session.resume("race_ws") is True # must not raise
assert session._model_binding is old_binding
assert session.model == "m-a"
def test_resume_construction_failure_logs_true_cause_keeps_binding(
self, tmp_db, monkeypatch, caplog
):
"""Logs the construction cause, not the unreachable-alias one."""
import logging
from turnstone.core.model_registry import ModelConfig, ModelRegistry
reg = ModelRegistry(
models={
"default": ModelConfig("default", "http://default/v1", "k", "m-default"),
"a": ModelConfig("a", "http://a/v1", "k", "m-a"),
},
default="default",
)
register_workstream("cons_ws")
save_message("cons_ws", "user", "hello")
save_workstream_config("cons_ws", {"model": "m-a", "model_alias": "a"})
binding = resolve_model_binding(reg, "default")
session = ChatSession(
client=binding.lane.client,
model=binding.lane.model,
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
registry=reg,
model_alias="default",
model_binding=binding,
)
old_binding = session._model_binding
def _boom(provider: str, **kwargs: object) -> object:
raise FileNotFoundError("/etc/ssl/missing-ca.pem")
monkeypatch.setattr(mr_module, "create_client", _boom)
with caplog.at_level(logging.WARNING):
assert session.resume("cons_ws") is True # must not raise
assert session._model_binding is old_binding
blob = " ".join(r.getMessage() for r in caplog.records)
assert "could not be constructed" in blob
assert "details in server log" in blob
assert "unreachable" not in blob
def test_init_does_not_clobber_existing_config(self, tmp_db):
"""ChatSession.__init__ must NOT overwrite existing
``workstream_config`` keys when constructing for an already-
persisted ws_id.
This is the fix for the rehydrate bug: ``SessionManager.open()``
builds a ChatSession with the persisted ws_id; the legacy
``__init__`` unconditionally called ``_save_config()`` which is
``INSERT OR REPLACE`` per-key — silently resetting model_alias,
temperature, reasoning_effort, max_tokens, skill, the persona
stamp, and instructions to the constructor defaults *before*
``resume()`` got a chance to read them back.
"""
client = MagicMock()
client.models.list.return_value.data = [MagicMock(id="test-model")]
ui = MagicMock()
ui.on_info = MagicMock()
ui.on_error = MagicMock()
ui.on_state_change = MagicMock()
ui.on_rename = MagicMock()
register_workstream("rehydrate_ws")
save_workstream_config(
"rehydrate_ws",
{
"model": "gpt-5-pro",
"model_alias": "gpt-5-pro",
"temperature": "0.2",
"reasoning_effort": "high",
"max_tokens": "8192",
"persona": "scribe",
"persona_prompt": "You are a scribe.",
"persona_tools": "[]",
"persona_mcp": "0",
"persona_memory": "0",
"instructions": "preserve me",
},
)
ChatSession(
client=client,
model="some-default-model",
ui=ui,
instructions=None,
temperature=0.7,
max_tokens=4096,
tool_timeout=30,
reasoning_effort="medium",
ws_id="rehydrate_ws",
)
loaded = load_workstream_config("rehydrate_ws")
assert loaded["model"] == "gpt-5-pro"
assert loaded["model_alias"] == "gpt-5-pro"
assert loaded["temperature"] == "0.2"
assert loaded["reasoning_effort"] == "high"
assert loaded["max_tokens"] == "8192"
assert loaded["persona"] == "scribe"
assert loaded["persona_tools"] == "[]"
assert loaded["persona_mcp"] == "0"
assert loaded["instructions"] == "preserve me"
def test_init_writes_config_on_fresh_create(self, tmp_db):
"""The opposite half of the contract: when no config row exists
yet, ``__init__`` must still persist the constructor's values so
a later resume can find them. This is the path that previously
worked — the fix must not break it."""
client = MagicMock()
client.models.list.return_value.data = [MagicMock(id="test-model")]
ui = MagicMock()
ui.on_info = MagicMock()
ui.on_error = MagicMock()
ui.on_state_change = MagicMock()
ui.on_rename = MagicMock()
# No save_workstream_config() before ChatSession() — this is
# the fresh-create path the SessionManager.create() flow takes.
register_workstream("fresh_ws")
assert load_workstream_config("fresh_ws") == {}
ChatSession(
client=client,
model="gpt-5-mini",
ui=ui,
instructions="be terse",
temperature=0.4,
max_tokens=2048,
tool_timeout=30,
reasoning_effort="low",
ws_id="fresh_ws",
)
loaded = load_workstream_config("fresh_ws")
assert loaded["model"] == "gpt-5-mini"
assert loaded["temperature"] == "0.4"
assert loaded["reasoning_effort"] == "low"
assert loaded["max_tokens"] == "2048"
assert loaded["instructions"] == "be terse"
# ── Prune workstreams ─────────────────────────────────────────────────
def _backdate_updated(ws_id: str) -> None:
"""Age a row past the orphan grace (and any retention cutoff)."""
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
conn.execute(
sa.text("UPDATE workstreams SET updated = '2020-01-01' WHERE ws_id = :ws"),
{"ws": ws_id},
)
conn.commit()
class TestPruneWorkstreams:
def test_orphan_removed(self, tmp_db):
"""An AGED empty workstream is pruned as an orphan (round-3 review:
eligibility now requires outliving the grace — see the fresh/named
twins below for the guards)."""
register_workstream("orphan")
_backdate_updated("orphan")
orphans, stale = prune_workstreams()
assert orphans == 1
assert list_workstreams_with_history() == []
def test_fresh_empty_workstream_survives_the_grace(self, tmp_db):
"""A just-registered empty workstream is a user mid-first-turn (its
rows may still be journal-held on a serving node another node's prune
cannot see) — never housekeeping debris. Round-3 review pin."""
register_workstream("fresh-empty")
orphans, stale = prune_workstreams()
assert (orphans, stale) == (0, 0)
assert get_storage().get_workstream("fresh-empty") is not None
def test_named_empty_workstream_never_pruned(self, tmp_db):
"""An aliased workstream is explicit user intent: excluded from the
orphan category regardless of age, mirroring the stale category's
alias exclusion. Round-3 review pin."""
register_workstream("named-empty")
set_workstream_alias("named-empty", "keep-me")
_backdate_updated("named-empty")
orphans, stale = prune_workstreams(retention_days=30)
assert (orphans, stale) == (0, 0)
assert get_storage().get_workstream("named-empty") is not None
def test_workstream_with_messages_kept(self, tmp_db):
"""Workstream with messages should not be pruned."""
register_workstream("active")
save_message("active", "user", "hello")
orphans, _stale = prune_workstreams()
assert orphans == 0
assert len(list_workstreams_with_history()) == 1
def test_stale_unnamed_removed(self, tmp_db):
"""Old unnamed workstream should be pruned by retention policy."""
register_workstream("old1")
save_message("old1", "user", "ancient message")
# Force the updated timestamp to the past so it looks stale
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
conn.execute(
sa.text("UPDATE workstreams SET updated = '2020-01-01' WHERE ws_id = 'old1'")
)
conn.commit()
_orphans, stale = prune_workstreams(retention_days=30)
assert stale == 1
def test_named_workstream_preserved(self, tmp_db):
"""Workstream with alias should be kept regardless of age."""
register_workstream("old2")
set_workstream_alias("old2", "important")
save_message("old2", "user", "old but named")
# Force old timestamp
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
conn.execute(
sa.text("UPDATE workstreams SET updated = '2020-01-01' WHERE ws_id = 'old2'")
)
conn.commit()
_orphans, stale = prune_workstreams(retention_days=30)
assert stale == 0
assert len(list_workstreams_with_history()) == 1
def test_fresh_unnamed_preserved(self, tmp_db):
"""Recent unnamed workstream should not be pruned."""
register_workstream("fresh")
save_message("fresh", "user", "just now")
_orphans, stale = prune_workstreams(retention_days=30)
assert stale == 0
assert len(list_workstreams_with_history()) == 1
def test_prune_removes_workstream_config(self, tmp_db):
"""Pruning orphan/stale workstreams should also remove their config rows."""
register_workstream("orphan_cfg")
_backdate_updated("orphan_cfg")
save_workstream_config("orphan_cfg", {"temperature": "0.5"})
register_workstream("stale_cfg")
save_message("stale_cfg", "user", "old")
save_workstream_config("stale_cfg", {"temperature": "0.9"})
engine = get_storage()._engine # noqa: SLF001
with engine.connect() as conn:
conn.execute(
sa.text("UPDATE workstreams SET updated = '2020-01-01' WHERE ws_id = 'stale_cfg'")
)
conn.commit()
# Both should have config before prune
assert load_workstream_config("orphan_cfg") == {"temperature": "0.5"}
assert load_workstream_config("stale_cfg") == {"temperature": "0.9"}
prune_workstreams(retention_days=30)
# Config rows should be cleaned up
assert load_workstream_config("orphan_cfg") == {}
assert load_workstream_config("stale_cfg") == {}
# ── Parallel tool exception isolation ────────────────────────────────
class TestParallelToolExceptionIsolation:
"""Bug #117: one tool raising should not kill the entire batch."""
def test_exception_in_one_tool_does_not_kill_batch(self, tmp_db, mock_openai_client):
from unittest.mock import patch
session = ChatSession(
client=mock_openai_client,
model="test-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
def succeed(item):
return item["call_id"], "ok"
def fail(item):
raise RuntimeError("boom")
items = [
{
"call_id": "c1",
"func_name": "bash",
"execute": succeed,
"needs_approval": False,
"header": "test",
"preview": "",
},
{
"call_id": "c2",
"func_name": "math",
"execute": fail,
"needs_approval": False,
"header": "test",
"preview": "",
},
]
tool_calls = [
{"id": "c1", "function": {"name": "bash", "arguments": "{}"}},
{"id": "c2", "function": {"name": "math", "arguments": "{}"}},
]
with (
patch.object(session, "_prepare_tool", side_effect=items),
patch.object(session, "_evaluate_intent"),
patch.object(session, "_emit_state"),
patch.object(session, "_init_system_messages"),
patch.object(session, "_check_cancelled"),
):
session.ui.approve_tools.return_value = (True, None)
results, _ = session._execute_tools(tool_calls)
assert results[0] == ("c1", "ok")
assert results[1][0] == "c2"
assert "Error executing math" in results[1][1]
assert "boom" in results[1][1]
# ── Web search tool gating ───────────────────────────────────────────
class TestWebSearchGating:
"""Bug #117: web_search should not be offered without a backend."""
def test_web_search_filtered_when_no_backend(self, tmp_db, mock_openai_client):
from unittest.mock import patch
from turnstone.core.providers._protocol import ModelCapabilities
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
caps = ModelCapabilities(supports_web_search=False)
with (
patch.object(session, "_get_capabilities", return_value=caps),
patch("turnstone.core.session.get_searxng_url", return_value=None),
):
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "web_search" not in names
def test_web_search_kept_when_searxng_configured(self, tmp_db, mock_openai_client):
from unittest.mock import patch
from turnstone.core.providers._protocol import ModelCapabilities
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
caps = ModelCapabilities(supports_web_search=False)
with (
patch.object(session, "_get_capabilities", return_value=caps),
patch("turnstone.core.session.get_searxng_url", return_value="http://searxng:8080"),
):
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "web_search" in names
def test_web_search_kept_when_native_support(self, tmp_db, mock_openai_client):
from unittest.mock import patch
from turnstone.core.providers._protocol import ModelCapabilities
session = ChatSession(
client=mock_openai_client,
model="gpt-5-search-api",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
caps = ModelCapabilities(supports_web_search=True)
with (
patch.object(session, "_get_capabilities", return_value=caps),
patch("turnstone.core.session.get_searxng_url", return_value=None),
):
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "web_search" in names
def test_resolve_search_client_config_store_precedence(self, tmp_db, mock_openai_client):
"""ConfigStore (admin Settings) is authoritative over the env/config
fallback: an explicit URL resolves a client; an explicit empty string
disables web search even when the env var would supply a URL; an unset
key falls through to env (storage -> toml -> env -> default)."""
from unittest.mock import patch
from turnstone.core.web_search import SearXNGClient
class _StubStore:
def __init__(self, values):
self._v = values
def stored_keys(self):
return frozenset(self._v)
def get(self, key, default=None):
return self._v.get(key, default)
def _session(values):
return ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
config_store=_StubStore(values),
)
# (a) explicit DB URL wins over a None env fallback
s_db = _session({"tools.searxng_url": "http://db-searx:8080"})
with patch("turnstone.core.session.get_searxng_url", return_value=None):
client = s_db._resolve_search_client()
assert isinstance(client, SearXNGClient)
assert client._base_url == "http://db-searx:8080"
# (b) explicit empty string disables, even though env would supply a URL
s_off = _session({"tools.searxng_url": ""})
with patch("turnstone.core.session.get_searxng_url", return_value="http://env-searx:8080"):
assert s_off._resolve_search_client() is None
# (c) unset key falls through to the env/config layer
s_env = _session({})
with patch("turnstone.core.session.get_searxng_url", return_value="http://env-searx:8080"):
client_env = s_env._resolve_search_client()
assert isinstance(client_env, SearXNGClient)
assert client_env._base_url == "http://env-searx:8080"
class TestMCPToolGating:
"""MCP tools should not be offered when no MCP servers provide them."""
def test_mcp_tools_filtered_without_mcp_client(self, tmp_db, mock_openai_client):
"""read_resource and use_prompt excluded when no MCP client."""
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
)
assert session._mcp_client is None
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "read_resource" not in names
assert "use_prompt" not in names
def test_read_resource_filtered_when_no_resources(self, tmp_db, mock_openai_client):
"""read_resource excluded when MCP client has no resources."""
mcp_client = MagicMock()
mcp_client.get_tools.return_value = []
# Phase 7b: gating uses ``*_count_for_user`` so the test mocks
# the per-user variant (the property remains for static-only
# admin paths). Returning 0 / 2 mirrors the prior contract.
mcp_client.resource_count_for_user.return_value = 0
mcp_client.prompt_count_for_user.return_value = 2
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
mcp_client=mcp_client,
)
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "read_resource" not in names
assert "use_prompt" in names
def test_use_prompt_filtered_when_no_prompts(self, tmp_db, mock_openai_client):
"""use_prompt excluded when MCP client has no prompts."""
mcp_client = MagicMock()
mcp_client.get_tools.return_value = []
mcp_client.resource_count_for_user.return_value = 3
mcp_client.prompt_count_for_user.return_value = 0
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
mcp_client=mcp_client,
)
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "use_prompt" not in names
assert "read_resource" in names
def test_mcp_tools_kept_when_servers_have_both(self, tmp_db, mock_openai_client):
"""Both tools present when MCP client has resources and prompts."""
mcp_client = MagicMock()
mcp_client.get_tools.return_value = []
mcp_client.resource_count_for_user.return_value = 1
mcp_client.prompt_count_for_user.return_value = 1
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
mcp_client=mcp_client,
)
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "read_resource" in names
assert "use_prompt" in names
def test_mcp_tools_filtered_with_tool_search_active(self, tmp_db, mock_openai_client):
"""Gating applies even when tool_search is active (client-side path)."""
mcp_client = MagicMock()
mcp_client.get_tools.return_value = []
mcp_client.resource_count_for_user.return_value = 0
mcp_client.prompt_count_for_user.return_value = 0
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
mcp_client=mcp_client,
tool_search="on",
)
assert session._tool_search is not None
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "read_resource" not in names
assert "use_prompt" not in names
def test_mcp_tools_filtered_with_native_tool_search(self, tmp_db, mock_openai_client):
"""Gating applies when provider handles tool search natively."""
from unittest.mock import patch
from turnstone.core.providers._protocol import ModelCapabilities
mcp_client = MagicMock()
mcp_client.get_tools.return_value = []
mcp_client.resource_count_for_user.return_value = 0
mcp_client.prompt_count_for_user.return_value = 0
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
mcp_client=mcp_client,
tool_search="on",
)
caps = ModelCapabilities(supports_tool_search=True)
with patch.object(session, "_get_capabilities", return_value=caps):
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "read_resource" not in names
assert "use_prompt" not in names
def test_pool_only_user_keeps_read_resource_and_use_prompt(self, tmp_db, mock_openai_client):
"""Phase 7b canary: a pool-only user (static catalog empty) still
sees ``read_resource`` and ``use_prompt`` because the gating
consults ``*_count_for_user`` (scope decision 0.2).
Drives ``resource_count = prompt_count = 0`` (the static-only
properties are zero) but ``*_count_for_user(uid) > 0`` because
the user has pool entries; the tools must remain visible.
"""
mcp_client = MagicMock()
mcp_client.get_tools.return_value = []
# Static catalog is empty; admin-style legacy properties say 0.
mcp_client.resource_count = 0
mcp_client.prompt_count = 0
# Per-user variant reports the user's pool entries.
mcp_client.resource_count_for_user.return_value = 2
mcp_client.prompt_count_for_user.return_value = 1
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
mcp_client=mcp_client,
user_id="pool-only-user",
)
tools = session._get_active_tools()
names = [t.get("function", {}).get("name") for t in tools]
assert "read_resource" in names
assert "use_prompt" in names
# Verify the per-user gate was actually consulted with the
# session's ``user_id`` (sanity-check on the wiring).
mcp_client.resource_count_for_user.assert_any_call("pool-only-user")
mcp_client.prompt_count_for_user.assert_any_call("pool-only-user")
class TestMCPActingUserBinding:
"""Per-user MCP credentials follow the acting user on shared workstreams.
The workstream owner is the fallback identity; an authenticated send
rebinds credential resolution (dispatch + catalogs + listeners) to the
sender. Prepared tool items pin the identity at prepare time so a
pending approval can't execute under a later sender's credentials.
"""
def _make(self, mock_openai_client, owner="alice"):
mcp_client = MagicMock()
mcp_client.get_tools.return_value = []
mcp_client.call_tool_sync.return_value = "ok"
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
mcp_client=mcp_client,
user_id=owner,
)
# Capture instead of persisting — same stub idiom as
# test_session_mcp_dispatch_error.
session._report_tool_result = MagicMock() # type: ignore[method-assign]
return session, mcp_client
def test_effective_identity_defaults_to_owner(self, tmp_db, mock_openai_client):
session, mcp_client = self._make(mock_openai_client)
assert session._mcp_effective_user_id == "alice"
item = session._prepare_mcp_tool("c1", "mcp__srv__tool", {})
session._exec_mcp_tool(item)
assert mcp_client.call_tool_sync.call_args.kwargs["user_id"] == "alice"
def test_bind_rebinds_dispatch_catalog_listeners_and_prime(self, tmp_db, mock_openai_client):
session, mcp_client = self._make(mock_openai_client)
mcp_client.reset_mock()
session.bind_acting_user("bob")
# Dispatch identity follows the acting user.
item = session._prepare_mcp_tool("c1", "mcp__srv__tool", {})
session._exec_mcp_tool(item)
assert mcp_client.call_tool_sync.call_args.kwargs["user_id"] == "bob"
# Listener registrations swapped from owner to acting user for
# all three catalog kinds — identity is the (user_id, callback)
# pair, so the remove must name the OLD uid and the add the new.
mcp_client.remove_listener.assert_called_once_with(session._mcp_refresh_cb, user_id="alice")
mcp_client.add_listener.assert_called_once_with(session._mcp_refresh_cb, user_id="bob")
mcp_client.remove_resource_listener.assert_called_once_with(
session._mcp_resource_cb, user_id="alice"
)
mcp_client.add_resource_listener.assert_called_once_with(
session._mcp_resource_cb, user_id="bob"
)
mcp_client.remove_prompt_listener.assert_called_once_with(
session._mcp_prompt_cb, user_id="alice"
)
mcp_client.add_prompt_listener.assert_called_once_with(
session._mcp_prompt_cb, user_id="bob"
)
# The acting user's oauth_user pools are warmed so their tools
# surface without a manual reconnect.
mcp_client.prime_user_pools.assert_called_once_with("bob")
# Merged tool list rebuilt under the new identity.
mcp_client.get_tools.assert_any_call(user_id="bob")
def test_status_snapshot_follows_acting_user(self, tmp_db, mock_openai_client):
"""The tool_search status snapshot scopes to the acting user, like
the get_tools call that builds the search corpus — owner-scoping
would render the owner's per-user pool state (including their
recorded discovery-failure text) into a non-owner participant's
search results."""
session, mcp_client = self._make(mock_openai_client)
mcp_client.get_all_server_status.return_value = {}
session.bind_acting_user("bob")
session._mcp_status_snapshot()
assert mcp_client.get_all_server_status.call_args.args == ("bob",)
def test_prepared_item_pins_identity_across_rebind(self, tmp_db, mock_openai_client):
session, mcp_client = self._make(mock_openai_client)
session.bind_acting_user("bob")
item = session._prepare_mcp_tool("c1", "mcp__srv__tool", {})
# A different user takes over the session while the item is
# pending approval — execution must stay under the requester.
session.bind_acting_user("carol")
session._exec_mcp_tool(item)
assert mcp_client.call_tool_sync.call_args.kwargs["user_id"] == "bob"
def test_resource_and_prompt_items_pin_identity(self, tmp_db, mock_openai_client):
session, mcp_client = self._make(mock_openai_client)
session.bind_acting_user("bob")
res_item = session._prepare_read_resource("c1", {"uri": "res://x"})
mcp_client.is_mcp_prompt.return_value = True
prompt_item = session._prepare_use_prompt("c2", {"name": "p"})
session.bind_acting_user("carol")
assert res_item["mcp_user_id"] == "bob"
assert prompt_item["mcp_user_id"] == "bob"
# And the prompt-existence gate consults the CURRENT effective
# identity (carol) for new preparations.
session._prepare_use_prompt("c3", {"name": "p"})
assert mcp_client.is_mcp_prompt.call_args.kwargs["user_id"] == "carol"
def test_bind_noops_on_empty_and_same_user(self, tmp_db, mock_openai_client):
session, mcp_client = self._make(mock_openai_client)
mcp_client.reset_mock()
session.bind_acting_user("")
session.bind_acting_user("alice") # same as owner
mcp_client.remove_listener.assert_not_called()
mcp_client.add_listener.assert_not_called()
mcp_client.prime_user_pools.assert_not_called()
assert session._mcp_effective_user_id == "alice"
def test_send_kwarg_binds_before_turn_starts(self, tmp_db, mock_openai_client):
import pytest
session, _mcp_client = self._make(mock_openai_client)
class _SentinelError(Exception):
pass
# ``bind_acting_user`` runs before ``_refresh_model_from_registry``
# at the top of send() — abort there to prove the ordering without
# driving the full agent loop.
session._refresh_model_from_registry = MagicMock( # type: ignore[method-assign]
side_effect=_SentinelError
)
with pytest.raises(_SentinelError):
session.send("hi", acting_user_id="bob")
assert session._acting_user_id == "bob"
def test_close_removes_listeners_under_rebound_identity(self, tmp_db, mock_openai_client):
session, mcp_client = self._make(mock_openai_client)
session.bind_acting_user("bob")
refresh_cb = session._mcp_refresh_cb
mcp_client.reset_mock()
session.close()
mcp_client.remove_listener.assert_called_once_with(refresh_cb, user_id="bob")
def test_bind_without_mcp_client_only_records(self, tmp_db, mock_openai_client):
session = ChatSession(
client=mock_openai_client,
model="local-model",
ui=MagicMock(),
instructions=None,
temperature=0.5,
max_tokens=1000,
tool_timeout=10,
user_id="alice",
)
session.bind_acting_user("bob")
assert session._mcp_effective_user_id == "bob"
Reference in New Issue View Git Blame Copy Permalink