diff --git a/.gitignore b/.gitignore index 3c8d3f49..9617c8e4 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,7 @@ build/ .venv/ venv/ .env +etc/ # Local compose overrides (e.g. run.sh's node-count limiter, bootstrap output) compose.override.yaml compose.override.yml diff --git a/tests/test_per_user_message_context.py b/tests/test_per_user_message_context.py new file mode 100644 index 00000000..43c35bd8 --- /dev/null +++ b/tests/test_per_user_message_context.py @@ -0,0 +1,314 @@ +"""Per-user message context (shared-workstream attribution). + +The context-identity layer atop upstream's acting-user credential fix: the model +must be TOLD who sent each user turn on a multi-user workstream, and that must +survive a worker rehydrating history from the DB. The sender is sourced from the +acting user (``_mcp_effective_user_id`` = the ``bind_acting_user`` initiator, +owner fallback); persistence rides ``conversations.meta`` (no migration). + +Covers: the ``_sender`` side-channel round-trip; DB replay routing; append-time +stamping from the acting user (and synthetic-turn exclusion); the wire-time +label injection gated on >1 distinct sender; and the shared-state detection + +one-time "has joined" note. +""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +from tests._session_helpers import make_session +from turnstone.core.session import _prefix_sender_label +from turnstone.core.storage._utils import reconstruct_turns +from turnstone.core.trajectory import Role, turn_from_dict, turn_to_dict + +# -- side-channel round-trip -------------------------------------------------- + + +def test_sender_round_trips_through_turn_dict(): + turn = turn_from_dict({"role": "user", "content": "hi", "_sender": "alice"}) + assert turn.meta.extra.get("sender") == "alice" + assert turn_to_dict(turn)["_sender"] == "alice" + + +def test_no_sender_leaves_no_key(): + turn = turn_from_dict({"role": "user", "content": "hi"}) + assert "sender" not in turn.meta.extra + assert "_sender" not in turn_to_dict(turn) + + +# -- reconstruct (DB replay) -------------------------------------------------- + + +def _user_row(row_id: int, content: str, meta: str | None): + # (id, role, content, tool_name, tc_id, provider_data, tool_calls, source, + # event_id, is_error, meta) + return (row_id, "user", content, None, None, None, None, None, None, False, meta) + + +def test_reconstruct_restores_user_sender_to_its_own_key(): + turns = reconstruct_turns([_user_row(1, "hello", json.dumps({"sender": "alice"}))], ws_id="ws1") + assert turns[0].meta.extra.get("sender") == "alice" + # Must NOT be misrouted into source_meta (that channel rides SYSTEM turns). + assert "source_meta" not in turns[0].meta.extra + + +def test_reconstruct_user_row_without_meta_has_no_sender(): + turns = reconstruct_turns([_user_row(1, "hello", None)], ws_id="ws1") + assert "sender" not in turns[0].meta.extra + + +# -- append stamps the sender from the ACTING user ---------------------------- + + +def test_append_stamps_and_persists_acting_user(): + s = make_session(user_id="owner") + s._acting_user_id = "alice" # a member drives this turn (bind_acting_user result) + with patch("turnstone.core.session.save_message", return_value=1) as sm: + s._append_user_turn("hello", ()) + assert sm.call_args.kwargs["meta"] == json.dumps({"sender": "alice"}) + assert s.messages[-1].meta.extra.get("sender") == "alice" + + +def test_append_owner_turn_stamps_owner(): + s = make_session(user_id="owner") # acting id empty -> effective = owner + with patch("turnstone.core.session.save_message", return_value=1) as sm: + s._append_user_turn("hello", ()) + assert sm.call_args.kwargs["meta"] == json.dumps({"sender": "owner"}) + + +def test_append_synthetic_turn_is_unstamped(): + s = make_session(user_id="owner") + s._acting_user_id = "alice" + with patch("turnstone.core.session.save_message", return_value=1) as sm: + s._append_user_turn("resuming", (), source="compaction_resume") + assert sm.call_args.kwargs["meta"] is None + assert "sender" not in s.messages[-1].meta.extra + + +# -- label injection (the model-visible half) --------------------------------- + + +def test_prefix_sender_label_string(): + assert _prefix_sender_label("do it", "alice") == "[message from alice]\ndo it" + + +def test_prefix_sender_label_multipart_folds_into_first_text(): + parts = [{"type": "text", "text": "look"}, {"type": "image", "attachment_id": "a1"}] + out = _prefix_sender_label(parts, "alice") + assert out[0]["text"] == "[message from alice]\nlook" + assert out[1] == {"type": "image", "attachment_id": "a1"} # untouched + assert parts[0]["text"] == "look" # input not mutated + + +def test_prefix_sender_label_attachment_only_inserts_leading_text(): + out = _prefix_sender_label([{"type": "image", "attachment_id": "a1"}], "alice") + assert out[0] == {"type": "text", "text": "[message from alice]"} + assert out[1] == {"type": "image", "attachment_id": "a1"} + + +def test_single_sender_not_labeled_same_ref(): + s = make_session(user_id="owner") + msgs = [ + {"role": "user", "content": "a", "_sender": "alice"}, + {"role": "user", "content": "b", "_sender": "alice"}, + ] + assert s._inject_sender_labels(msgs) is msgs # allocation-free common case + + +def test_shared_state_labels_even_when_slice_has_single_sender(): + # Compaction can narrow the wire slice to one participant's turns. On a + # known-shared workstream we must still label (the >1-sender count heuristic + # alone would skip and let the model misattribute to the owner). + s = make_session(user_id="owner") + s._shared_workstream = True + msgs = [{"role": "user", "content": "only alice remains", "_sender": "alice"}] + with patch("turnstone.core.session.get_storage", return_value=None): + out = s._inject_sender_labels(msgs) + assert out is not msgs + assert out[0]["content"] == "[message from alice]\nonly alice remains" + + +def test_shared_labels_every_sender_turn(): + # No storage -> _resolve_display_name falls back to the raw id, so labels + # carry the id here (username resolution is covered separately below). + s = make_session(user_id="owner") + msgs = [ + {"role": "user", "content": "from owner", "_sender": "owner"}, + {"role": "assistant", "content": "hi"}, + {"role": "user", "content": "from member", "_sender": "alice"}, + ] + with patch("turnstone.core.session.get_storage", return_value=None): + out = s._inject_sender_labels(msgs) + assert out is not msgs + assert out[0]["content"] == "[message from owner]\nfrom owner" + assert out[2]["content"] == "[message from alice]\nfrom member" + assert out[1]["content"] == "hi" # assistant untouched + assert msgs[0]["content"] == "from owner" # canonical input untouched + + +def test_shared_leaves_synthetic_unlabeled(): + s = make_session(user_id="owner") + msgs = [ + {"role": "user", "content": "hi", "_sender": "owner"}, + {"role": "user", "content": "hey", "_sender": "alice"}, + {"role": "user", "content": "", "_source": "wake"}, # synthetic: no _sender + ] + with patch("turnstone.core.session.get_storage", return_value=None): + out = s._inject_sender_labels(msgs) + assert out[2]["content"] == "" # untouched -> still drops as an empty wire turn + + +# -- display-name resolution (senders read as usernames, not id hashes) ------- + + +def test_resolve_display_name_owner_uses_session_username(): + s = make_session(user_id="owner", username="owner@example") + assert s._resolve_display_name("owner") == "owner@example" + + +def test_resolve_display_name_others_via_storage_and_caches(): + s = make_session(user_id="owner") + fake = MagicMock() + fake.get_user.return_value = {"username": "alice@example", "display_name": "Alice"} + with patch("turnstone.core.session.get_storage", return_value=fake): + assert s._resolve_display_name("alice-id") == "alice@example" + assert s._resolve_display_name("alice-id") == "alice@example" # cache hit + fake.get_user.assert_called_once() # second lookup served from cache + + +def test_resolve_display_name_falls_back_to_id_when_unknown(): + s = make_session(user_id="owner") + fake = MagicMock() + fake.get_user.return_value = None + with patch("turnstone.core.session.get_storage", return_value=fake): + assert s._resolve_display_name("ghost-id") == "ghost-id" + + +def test_resolve_display_name_retries_after_transient_storage_error(): + # A storage error must NOT be cached: it falls back to the raw id for this + # call but a later call retries and resolves, rather than pinning the id. + s = make_session(user_id="owner") + fake = MagicMock() + fake.get_user.side_effect = [RuntimeError("storage down"), {"username": "alice@example"}] + with patch("turnstone.core.session.get_storage", return_value=fake): + assert s._resolve_display_name("alice-id") == "alice-id" # error -> raw id, uncached + assert s._resolve_display_name("alice-id") == "alice@example" # retried, resolved + assert fake.get_user.call_count == 2 + + +def test_labels_render_resolved_usernames(): + s = make_session(user_id="owner") + fake = MagicMock() + fake.get_user.side_effect = lambda uid: { + "owner": {"username": "owner@example"}, + "alice-id": {"username": "alice@example"}, + }.get(uid) + msgs = [ + {"role": "user", "content": "a", "_sender": "owner"}, + {"role": "user", "content": "b", "_sender": "alice-id"}, + ] + with patch("turnstone.core.session.get_storage", return_value=fake): + out = s._inject_sender_labels(msgs) + assert out[0]["content"] == "[message from owner@example]\na" + assert out[1]["content"] == "[message from alice@example]\nb" + + +# -- shared-state detection + join note --------------------------------------- + + +def test_recompute_shared_state_from_history(): + s = make_session(user_id="owner") + s.messages.append(turn_from_dict({"role": "user", "content": "a", "_sender": "owner"})) + s._recompute_shared_state() + assert s._shared_workstream is False # owner alone is not shared + s.messages.append(turn_from_dict({"role": "user", "content": "b", "_sender": "alice"})) + s._recompute_shared_state() + assert s._shared_workstream is True + assert s._known_senders == {"owner", "alice"} + + +def test_new_participant_flips_shared_and_emits_join_note_once(): + s = make_session(user_id="owner") + s._known_senders = {"owner"} + with ( + patch.object(s, "_init_system_messages") as recompose, + patch("turnstone.core.session.get_storage", return_value=None), + ): + s._maybe_note_new_participant("alice") + assert s._shared_workstream is True + recompose.assert_called_once() # banner recomposed on the shared transition + assert s.messages[-1].role is Role.SYSTEM + assert s.messages[-1].source == "participant_joined" + n = len(s.messages) + # owner and a repeat participant are no-ops (no duplicate join note) + s._maybe_note_new_participant("owner") + s._maybe_note_new_participant("alice") + assert len(s.messages) == n + + +def test_owner_only_never_shared(): + s = make_session(user_id="owner") + with patch.object(s, "_init_system_messages") as recompose: + s._maybe_note_new_participant("owner") + assert s._shared_workstream is False + recompose.assert_not_called() + + +# -- Session Context banner (shared vs single-user) --------------------------- + + +def test_shared_banner_declares_participants_and_tool_credentials(): + from turnstone.prompts import SessionContext, WorkstreamKind, _build_context + + shared = _build_context( + SessionContext(current_datetime="t", timezone="UTC", username="owner@x", shared=True), + WorkstreamKind.INTERACTIVE, + ) + solo = _build_context( + SessionContext(current_datetime="t", timezone="UTC", username="owner@x", shared=False), + WorkstreamKind.INTERACTIVE, + ) + # shared: multi-user framing + per-message tags + the tool-credential rule + assert "SHARED workstream" in shared + assert "message from" in shared + assert "credentials of the participant" in shared + assert "different results" in shared.lower() or "DIFFERENT results" in shared + # single-user: unchanged simple owner line, no shared framing + assert "- **User:** owner@x" in solo + assert "SHARED" not in solo + + +# -- workstream / project identifiers in context ------------------------------ + + +def test_context_surfaces_workstream_and_project_ids(): + from turnstone.prompts import SessionContext, WorkstreamKind, _build_context + + out = _build_context( + SessionContext( + current_datetime="t", + timezone="UTC", + username="owner@x", + project="My Project", + project_id="proj-123", + ws_id="ws-abc", + ), + WorkstreamKind.INTERACTIVE, + ) + assert "- **Workstream ID:** ws-abc" in out + # project renders both its display name and its stable id + assert "My Project" in out + assert "proj-123" in out + + +def test_context_omits_ids_when_absent(): + from turnstone.prompts import SessionContext, WorkstreamKind, _build_context + + out = _build_context( + SessionContext(current_datetime="t", timezone="UTC", username="owner@x"), + WorkstreamKind.INTERACTIVE, + ) + # no ws_id line and no project line at all when neither is set + assert "Workstream ID" not in out + assert "**Project:**" not in out diff --git a/tests/test_session_attachments.py b/tests/test_session_attachments.py index ee2f4307..21bcdd89 100644 --- a/tests/test_session_attachments.py +++ b/tests/test_session_attachments.py @@ -60,16 +60,30 @@ def _run_send(session: ChatSession, text: str, attachments=None) -> None: raise +def _assert_plain_text_turn(d: dict) -> None: + """A plain-text send (no attachments) must NOT be coerced into the + multipart/attachment shape: ``content`` stays the plain string and no + ``_attachments_meta`` is emitted. The per-user-context feature stamps every + genuine user turn with a wire-invisible ``_sender`` attribution key (a + leading-underscore side channel, stripped by ``sanitize_messages`` before + the model call), so it may be present alongside role/content — that is the + only addition tolerated here.""" + assert d["role"] == "user" + assert d["content"] == "hello" # plain string, not a multipart list + assert "_attachments_meta" not in d + assert set(d) <= {"role", "content", "_sender"} + + class TestPlainTextUnchanged: def test_no_attachments_stores_string_content(self, tmp_db, mock_openai_client): s = _make_session(mock_openai_client) _run_send(s, "hello") - assert turn_to_dict(s.messages[-1]) == {"role": "user", "content": "hello"} + _assert_plain_text_turn(turn_to_dict(s.messages[-1])) def test_empty_attachments_list_stores_string_content(self, tmp_db, mock_openai_client): s = _make_session(mock_openai_client) _run_send(s, "hello", attachments=[]) - assert turn_to_dict(s.messages[-1]) == {"role": "user", "content": "hello"} + _assert_plain_text_turn(turn_to_dict(s.messages[-1])) class TestMultipartBuild: diff --git a/tests/test_skills_tool.py b/tests/test_skills_tool.py index 82ee0284..046dc82f 100644 --- a/tests/test_skills_tool.py +++ b/tests/test_skills_tool.py @@ -1369,6 +1369,10 @@ class TestSkillCatalogDisclosure: session._memory_config = MagicMock() session._memory_config.fetch_limit = 0 session._user_id = "test-user" + # _init_system_messages -> _recompute_shared_state reads the session + # owner (_mcp_user_id) to decide shared-workstream framing; __init__ + # normally sets it from user_id, so seed it here for the __new__ build. + session._mcp_user_id = "test-user" with ( patch( diff --git a/turnstone/core/metacognition.py b/turnstone/core/metacognition.py index 28488b4b..3be9e7be 100644 --- a/turnstone/core/metacognition.py +++ b/turnstone/core/metacognition.py @@ -146,6 +146,14 @@ _NUDGE_MAP: dict[str, str] = { # consumers recognise the type. "idle_children": "", "watch_triggered": "", + # participant_joined likewise carries no static body — the per-fire text + # (" has joined this shared workstream…") is composed by its producer + # (``ChatSession._maybe_note_new_participant``) and emitted via + # ``_append_system_turn``, never through :func:`format_nudge`. The entry + # exists only to keep this map mirroring ``tool_advisory.SYSTEM_TURN_SOURCES`` + # (enforced by ``test_vocabulary_mirrors_nudge_map_both_directions``); nothing + # calls ``should_nudge("participant_joined", …)`` so it never auto-fires. + "participant_joined": "", } diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 4e2b77e7..ce71a472 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -271,6 +271,29 @@ _IMAGE_EXTENSIONS: frozenset[str] = frozenset( _IMAGE_SIZE_CAP = _ATTACH_IMAGE_SIZE_CAP +def _prefix_sender_label(content: Any, sender: str) -> Any: + """Return *content* with a ``[message from ]`` header prepended. + + Handles both the plain-string user content and the multipart list shape + (text + attachment parts): the header is folded into the first text part, or + inserted as a leading text part when the content is attachment-only. Returns + a new object; the input is never mutated (the caller works on a transient + wire copy, not the canonical ``self.messages``).""" + label = f"[message from {sender}]\n" + if isinstance(content, str): + return label + content + if isinstance(content, list): + out = list(content) + for i, part in enumerate(out): + if isinstance(part, dict) and part.get("type") == "text": + np = dict(part) + np["text"] = label + str(np.get("text", "")) + out[i] = np + return out + return [{"type": "text", "text": label.rstrip("\n")}, *out] + return content + + def _encode_image_data_uri(raw: bytes, mime: str) -> str: """Wrap raw image bytes as a ``data:{mime};base64,...`` URI.""" b64 = base64.b64encode(raw).decode("ascii") @@ -1187,6 +1210,18 @@ class ChatSession: # the ``(user_id, callback)`` pair). self._acting_user_id: str = "" self._mcp_listener_user_id: str | None = user_id or None + # Shared-workstream context state (context-identity layer, atop the + # acting-user credential fix above): the model must be TOLD when more + # than one human is in the room. ``_shared_workstream`` flips True once a + # non-owner sender appears (live send OR rehydrated history) and drives + # the ``## Session Context`` banner; ``_known_senders`` tracks who has + # spoken so a first-time participant gets a one-time "has joined" note. + self._shared_workstream: bool = False + self._known_senders: set[str] = set() + # user_id -> display username cache for shared-workstream labels / join + # notes, so senders read as usernames (like the owner banner) not raw + # id hashes. Resolved lazily via storage; a handful of entries per ws. + self._sender_name_cache: dict[str, str] = {} self._username = username self._client_type = client_type # Whether the user is online to complete an in-flight OAuth @@ -3101,11 +3136,17 @@ class ChatSession: # on every turn that crossed a minute boundary. Hour-precision still # gives the model time-of-day awareness without paying for a full # prefix recompute every ~60 seconds. + # Refresh shared-workstream state so the banner matches the current + # participant set (fresh compose or rehydrated multi-user history). + self._recompute_shared_state() ctx = SessionContext( current_datetime=now.strftime("%Y-%m-%dT%H:00"), timezone=now.tzname() or "UTC", username=self._username or self._user_id or "unknown", project=self._project_name, + shared=self._shared_workstream, + ws_id=self._ws_id, + project_id=self._project_id, ) composed = compose_system_message( client_type=self._client_type, @@ -3578,6 +3619,127 @@ class ChatSession: ), } + def _resolve_display_name(self, user_id: str) -> str: + """Resolve a user_id to its display username for shared-workstream + labels / join notes — the same *kind* of identity the owner gets in the + Session Context banner (``self._username``), rather than a raw id hash. + + The owner short-circuits to ``self._username`` (already known, matches + the banner exactly). Others are looked up once via storage and cached on + the session; a lookup miss / no user row falls back to the raw id so the + label degrades gracefully rather than disappearing.""" + if not user_id: + return "" + if user_id == self._mcp_user_id and self._username: + return self._username + cached = self._sender_name_cache.get(user_id) + if cached is not None: + return cached + name = user_id + try: + storage = get_storage() + if storage: + row = storage.get_user(user_id) + if row: + name = row.get("username") or row.get("display_name") or user_id + # Cache hits and definite misses (row is None). A transient + # storage error, by contrast, skips the cache and falls through + # to return the raw id, so a later call can retry rather than + # pinning the sender to their id for the session's lifetime. + self._sender_name_cache[user_id] = name + except Exception: + log.debug("display-name lookup failed for user=%s", user_id, exc_info=True) + return name + + def _recompute_shared_state(self) -> None: + """Recompute shared-workstream state from the current trajectory. + + Scans user turns for recorded senders (the ``meta.extra["sender"]`` + stamped by :meth:`_append_user_turn`). The workstream is *shared* once a + sender other than the owner (``_mcp_user_id``) has spoken. Called at + system-prompt (re)composition so the ``## Session Context`` banner + reflects reality on a fresh compose AND on a worker rehydrating an + already-multi-user workstream from history.""" + owner = (self._mcp_user_id or "").strip() + senders = { + s + for t in self.messages + if t.role is Role.USER and (s := (t.meta.extra.get("sender") or "").strip()) + } + self._known_senders = senders + self._shared_workstream = any(s != owner for s in senders) + + def _maybe_note_new_participant(self, sender_user_id: str | None) -> None: + """Announce a first-time non-owner sender and flip the ws to shared. + + Called from :meth:`send` right after the user turn is appended, with the + turn's acting user. The first non-owner sender recomposes the system + prefix so the banner switches to multi-user framing; every first-time + participant (2nd, 3rd, …) also gets a one-time ``participant_joined`` + system turn the model sees on that same turn — the "bob has joined the + chat" signal, since we can't know a participant exists until they speak. + The owner and repeat senders are no-ops.""" + owner = (self._mcp_user_id or "").strip() + s = (sender_user_id or "").strip() + if not s or s == owner or s in self._known_senders: + return + was_shared = self._shared_workstream + self._known_senders.add(s) + self._shared_workstream = True + if not was_shared: + # First non-owner sender: recompose so the banner is multi-user. + # (_recompute_shared_state re-derives the set from history, which now + # includes this sender's just-appended turn — consistent.) + self._init_system_messages() + name = self._resolve_display_name(s) + self._append_system_turn( + "participant_joined", + f"{name} has joined this shared workstream. Their messages are tagged " + f"`[message from {name}]` — attribute them to this sender, not the owner.", + ) + + def _inject_sender_labels(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Fold per-message sender attribution into user turns for the wire. + + The context-layer half of per-user identity: on a genuinely multi-user + (shared) workstream the model must be told *who* sent each turn, or it + conflates every participant. Only user turns that recorded a real sender + (the ``_sender`` side channel from :meth:`_append_user_turn`) are + considered — synthetic turns (wake, compaction-resume, advisory) carry + none and stay unlabeled. + + "Shared" is authoritative session state (``_shared_workstream`` — flipped + once a non-owner speaks and re-derived when a worker rehydrates history). + The per-slice sender count is only a fallback for when that state is still + unset/uncomputed: keying off the count alone would drop labels whenever + compaction narrows the retained wire slice to a single participant on a + known-shared workstream, reintroducing the misattribution the banner tells + the model these labels prevent. A single-user workstream returns the input + unchanged (same object reference — the allocation-free common case + ``_prepare_wire_messages`` relies on). The label rides the model-visible + ``content``; the wire-invisible ``_sender`` key is stripped downstream by + ``sanitize_messages``. ``self.messages`` is never mutated.""" + if not self._shared_workstream: + senders = { + s + for m in messages + if m.get("role") == "user" and (s := (m.get("_sender") or "").strip()) + } + if len(senders) <= 1: + return messages + out: list[dict[str, Any]] = [] + for m in messages: + sender = (m.get("_sender") or "").strip() if m.get("role") == "user" else "" + if sender: + nm = dict(m) + nm["content"] = _prefix_sender_label( + m.get("content"), self._resolve_display_name(sender) + ) + out.append(nm) + else: + out.append(m) + return out + def _prepare_wire_messages( self, messages: list[dict[str, Any]], @@ -3618,6 +3780,10 @@ class ChatSession: # threads the dicts ``_full_messages`` already produced straight through — # no Turn round-trip per send (``self.messages`` stays the canonical Turn # truth; only this transient wire copy is dict-native). + # Per-user attribution first: label user turns by sender on shared + # workstreams (no-op / same-ref on single-user) BEFORE folding so the + # label is part of the content the fold + repair passes carry through. + messages = self._inject_sender_labels(messages) folded = messages if self._provider is not None: folded = fold_system_turns( @@ -4304,6 +4470,17 @@ class ChatSession: # auto-resume): marks the turn for audit / replay / UI so it isn't # mistaken for real user input. Stripped at the sanitize boundary. user_msg["_source"] = source + # Per-message sender identity for shared-workstream attribution (the + # context-identity layer atop upstream's acting-user credential fix). + # Stamp genuine user turns with the ACTING user — the turn initiator + # bound by ``bind_acting_user`` (owner fallback for CLI/eval/internal). + # Synthetic turns (wake, compaction-resume, advisory) carry a ``_source`` + # / ``from_wake`` and stay unstamped so they never get a speaker label. + # Rides the wire-invisible ``_sender`` side channel and, below, the + # persisted ``meta`` column so history replay re-attributes correctly. + sender = "" if (from_wake or source) else (self._mcp_effective_user_id or "").strip() + if sender: + user_msg["_sender"] = sender if attachments: # Sibling metadata so live history replay has the same shape # as reloaded-from-DB (filenames are not recoverable from an @@ -4335,12 +4512,18 @@ class ChatSession: # longer ride this row — the caller appends them as first-class # ``system`` turns AFTER this user turn (uniform attach rule). source = user_msg.get("_source") + # Persist the sender in the row's ``meta`` JSON (no schema change — the + # column already carries opaque per-row metadata). ``reconstruct_turns`` + # restores it to ``Turn.meta.extra["sender"]`` so a worker rehydrating a + # shared workstream re-attributes each user turn. + meta_json = json.dumps({"sender": sender}) if sender else None message_id = save_message( self._ws_id, "user", user_input, source=source if isinstance(source, str) and source else None, event_id=self._ui_event_id(), + meta=meta_json, ) if attachments and message_id: self._persist_attachment_refs(message_id, attachments) @@ -4665,6 +4848,13 @@ class ChatSession: self._queue_user_advisory(*nudge) self._append_user_turn(user_input, attachments or (), send_id=send_id, from_wake=from_wake) + # Context-identity: if a new (non-owner) participant just spoke, flip the + # workstream to shared framing (banner recompose) and drop a one-time + # "has joined" note so the model learns a second human exists — it can't + # know until they send a message. Sourced from the acting user bound + # above (empty/owner for CLI/eval/internal turns → no-op). + if not from_wake: + self._maybe_note_new_participant(self._mcp_effective_user_id) # Drained user-channel nudges become first-class ``system`` turns # appended AFTER the user turn (uniform attach rule), replacing the # legacy per-message ``_reminders`` side-channel splice. diff --git a/turnstone/core/storage/_utils.py b/turnstone/core/storage/_utils.py index 6f0ad51f..58de9903 100644 --- a/turnstone/core/storage/_utils.py +++ b/turnstone/core/storage/_utils.py @@ -918,6 +918,13 @@ def reconstruct_turns( # rows (bare source_meta dict, no effect_status key) fall through. if role == "tool" and "effect_status" in raw_meta: meta.extra["effect_status"] = raw_meta["effect_status"] + elif role == "user" and "sender" in raw_meta: + # Per-message sender identity (shared-workstream attribution). + # A USER row's meta blob carries only ``{"sender": ...}`` — route + # it to its own key so history replay re-attributes each turn to + # the human who sent it (source_meta rides SYSTEM turns, never + # user turns, so there is no collision). + meta.extra["sender"] = raw_meta["sender"] else: meta.extra["source_meta"] = raw_meta src = str(source) if source else None diff --git a/turnstone/core/tool_advisory.py b/turnstone/core/tool_advisory.py index 2bf2506a..f5753a67 100644 --- a/turnstone/core/tool_advisory.py +++ b/turnstone/core/tool_advisory.py @@ -125,6 +125,7 @@ SYSTEM_TURN_SOURCES: Final = frozenset( "compaction_pending", "idle_children", "watch_triggered", + "participant_joined", } ) diff --git a/turnstone/core/trajectory.py b/turnstone/core/trajectory.py index d7c2b3c0..3855fe03 100644 --- a/turnstone/core/trajectory.py +++ b/turnstone/core/trajectory.py @@ -305,6 +305,14 @@ def turn_from_dict(msg: dict[str, Any]) -> Turn: es = msg.get("_effect_status") if es: meta.extra["effect_status"] = es + # Per-message sender identity for shared workstreams (who actually sent this + # user turn). Wire-invisible side channel — folded into the model-visible + # content at :meth:`ChatSession._prepare_wire_messages` only when the + # workstream is multi-user; the ``_``-prefixed key itself never reaches the + # provider (stripped by ``sanitize_messages``). + sndr = msg.get("_sender") + if sndr: + meta.extra["sender"] = sndr return Turn( role=role, @@ -351,6 +359,9 @@ def turn_to_dict(turn: Turn) -> dict[str, Any]: es = turn.meta.extra.get("effect_status") if es: msg["_effect_status"] = es + sndr = turn.meta.extra.get("sender") + if sndr: + msg["_sender"] = sndr return msg diff --git a/turnstone/prompts/__init__.py b/turnstone/prompts/__init__.py index 0d2a1fc2..382fcfc3 100644 --- a/turnstone/prompts/__init__.py +++ b/turnstone/prompts/__init__.py @@ -58,6 +58,9 @@ class SessionContext: timezone: str # system tz abbreviation, required username: str # users.username, required project: str = "" # attached project name, rendered only when the ws has one + shared: bool = False # True once >1 distinct human sends into the workstream + ws_id: str = "" # this workstream's stable id — lets the model name "this workstream" + project_id: str = "" # attached project's stable id (human-readable name is ``project``) # File-based policy-to-tool gating (defaults). @@ -75,13 +78,49 @@ _ENV_MAP: dict[ClientType, str] = { def _build_context(ctx: SessionContext, kind: WorkstreamKind) -> str: - """Build the CONTEXT module from session variables.""" - project_line = f"- **Project:** {ctx.project}\n" if ctx.project else "" + """Build the CONTEXT module from session variables. + + In a **shared** workstream more than one person sends messages, so a single + ``- **User:**`` line would mislead the model into attributing every turn to + the owner. We instead name the owner and declare the workstream multi-user, + pointing the model at the per-message ``[message from ]`` tags (added at + :meth:`ChatSession._prepare_wire_messages`) as the authoritative per-turn + identity. The single-user line is unchanged for the common case. + + The workstream id and project id are surfaced (when present) so the model + can refer to *this* workstream/project by its stable handle — e.g. when + registering an out-of-band callback (an alert that should feed follow-ups + back into this same workstream) rather than only its display name. Both + are stable for the session, so they don't perturb the cached prompt prefix. + """ + ws_line = f"- **Workstream ID:** {ctx.ws_id}\n" if ctx.ws_id else "" + if ctx.project: + project_display = ctx.project + if ctx.project_id: + project_display += f" (id: {ctx.project_id})" + project_line = f"- **Project:** {project_display}\n" + else: + project_line = "" + if ctx.shared: + who_lines = ( + f"- **Owner:** {ctx.username}\n" + "- **Participants:** SHARED workstream — more than one person sends messages " + "here. Each user turn is prefixed `[message from ]` with the " + "identity of whoever sent it; attribute requests and statements to that sender " + "and do NOT assume a single user. Tool / MCP calls execute under the " + "credentials of the participant who initiated the current turn (usually the " + "sender, not the owner), so the SAME tool can legitimately return DIFFERENT " + "results for different senders — that is expected, not an error or " + "inconsistency.\n" + ) + else: + who_lines = f"- **User:** {ctx.username}\n" return ( "## Session Context\n" "\n" f"- **Current date/time:** {ctx.current_datetime} ({ctx.timezone})\n" - f"- **User:** {ctx.username}\n" + f"{ws_line}" + f"{who_lines}" f"{project_line}" f"- **Session kind:** {kind.value}" )