Multi-user chat context clarification and tool improvements (#750)

* multiuser chat fixes for identity clarity and obo oauth token selection during tool calls

* added some missing context to the session so that the llm would know what session/project to reference in tool calls

* updated to address copilots issues and excluded a local config folder

* I think this resolves the cicd failures

---------

Co-authored-by: pow3rtool <root@pow3rtools>

.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

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

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:

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(

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
+    # ("<name> 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": "",
 }
 
 

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 <sender>]`` 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.

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

turnstone/core/tool_advisory.py

@@ -125,6 +125,7 @@ SYSTEM_TURN_SOURCES: Final = frozenset(
         "compaction_pending",
         "idle_children",
         "watch_triggered",
+        "participant_joined",
     }
 )
 

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
 
 

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 <id>]`` 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 <participant>]` 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}"
     )