mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
feat(memory): durable per-user coordinator scope + anonymous-coordinator guard
The coordinator memory scope was keyed by the session's ws_id, so every new coordinator session started with an empty namespace and its rows were orphaned on close — coordinator memory never actually persisted. Re-key the scope to the coordinator's creator user_id: one durable orchestration namespace per user, shared by all of that user's coordinator sessions (concurrent ones included; upsert-by-name is the collision rule). The child-containment threat model is unchanged: the gate is session KIND — children are always interactive and share the parent's user_id, so _validate_scope rejects them before scope resolution, and the REST memories API still rejects the coordinator scope outright. The implicit visibility lane now also fails closed on an empty scope_id to match the explicit search/list lanes (the storage helpers treat a falsy scope_id as 'no scope_id filter', which would have read every user's rows). Anonymous coordinators are no longer constructible: ChatSession refuses kind=COORDINATOR with an empty user_id at the constructor — the single choke point covering create, rehydration of legacy rows (surfaced by the open handler as a 503 with remediation text), and any future host — and the console no longer masks an empty uid as a phantom 'system' principal when minting coordinator JWTs, per CoordinatorTokenManager's documented 'sub = the real creator user_id' contract. Migration 061 carries existing coordinator rows across: rows whose owning workstream is gone or ownerless are deleted (unreachable under user keying), same-name collisions within a user keep the newest updated row (memory_id tiebreak), and survivors re-key to the owner's user_id.
This commit is contained in:
Patrick Buckley
@@ -77,7 +77,7 @@ or MCP config can do adds to it. Current members:
|
||||
| `delete_workstream` | wind-down | Hard-delete one child. Requires approval. |
|
||||
| `list_nodes` | discover | Enumerate live cluster nodes + capabilities. |
|
||||
| `skills` (action=find) | discover | Browse the skill catalog; opt-in `kind` filter narrows by audience. |
|
||||
| `memory` | persist | Orchestration scratchpad keyed by the `coordinator` scope. |
|
||||
| `memory` | persist | Durable orchestration memory (`coordinator` scope, per-user — survives across coordinator sessions). |
|
||||
| `notify` | broadcast | Post a status update to a human channel at a narrative beat. |
|
||||
| `tasks` | plan | Orchestrator-only scratchpad. Children don't see it. |
|
||||
|
||||
|
||||
+30
-5
@@ -26,15 +26,40 @@ Each memory has three dimensions:
|
||||
|
||||
### Memory scopes
|
||||
|
||||
| Scope | Visibility |
|
||||
|--------------|-----------------------------------------------------------|
|
||||
| `global` | Visible to all workstreams and users |
|
||||
| `workstream` | Visible only within the originating workstream |
|
||||
| `user` | Follows the authenticated user across workstreams |
|
||||
| Scope | Visibility |
|
||||
|---------------|-----------------------------------------------------------------|
|
||||
| `global` | Visible to all workstreams and users |
|
||||
| `workstream` | Visible only within the originating workstream |
|
||||
| `user` | Follows the authenticated user across workstreams |
|
||||
| `coordinator` | Coordinator sessions only; follows the user across coordinators |
|
||||
|
||||
A memory's identity is the tuple `(name, scope, scope_id)`. Saving a memory
|
||||
with the same identity upserts -- updating content while preserving the ID.
|
||||
|
||||
### Coordinator scope
|
||||
|
||||
Coordinator sessions are isolated to a single scope: `coordinator`, keyed by
|
||||
the coordinator's creator `user_id`. It is durable -- every coordinator
|
||||
session the same user runs (including concurrent ones) shares one
|
||||
orchestration namespace, so procedures and lessons survive close/reopen.
|
||||
|
||||
Isolation is bidirectional and enforced by session kind, not by secrecy of
|
||||
the scope id:
|
||||
|
||||
- A coordinator session can read and write **only** `coordinator`-scope rows.
|
||||
It never sees `global`/`workstream`/`user` memories, so content written by
|
||||
interactive sessions (which routinely ingest untrusted MCP/attachment
|
||||
output) cannot reach a coordinator's system message.
|
||||
- Interactive sessions -- including a coordinator's own children, which share
|
||||
its `user_id` -- are rejected from the `coordinator` scope on every memory
|
||||
action. Children cannot plant rows the parent coordinator would read.
|
||||
- The REST memory API (`/v1/api/memories`) does not accept the `coordinator`
|
||||
scope at all; the scope is written exclusively through a coordinator
|
||||
session's own memory tool.
|
||||
|
||||
Coordinator sessions require an authenticated user identity -- an anonymous
|
||||
coordinator cannot be constructed, so the scope id is always a real user.
|
||||
|
||||
### BM25 relevance injection
|
||||
|
||||
On every conversation turn, the system:
|
||||
|
||||
@@ -630,6 +630,7 @@ def test_prepare_fails_cleanly_when_coord_client_missing(monkeypatch):
|
||||
tool_timeout=30,
|
||||
context_window=16384,
|
||||
ws_id="coord-1",
|
||||
user_id="user-1", # the constructor refuses anonymous coordinators
|
||||
kind="coordinator",
|
||||
coord_client=None,
|
||||
)
|
||||
|
||||
@@ -452,7 +452,9 @@ class TestCompositionCandidateSelection:
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
scopes = coord._visible_scopes()
|
||||
assert scopes == [("coordinator", "coord-1")]
|
||||
# Keyed by the creator user_id (durable per-user namespace),
|
||||
# not the session's ws_id.
|
||||
assert scopes == [("coordinator", "user-1")]
|
||||
# And: search uses those same scopes (no global/user fan-in)
|
||||
coord.messages = turns_from_dicts([{"role": "user", "content": "anything"}])
|
||||
with patch(
|
||||
@@ -462,7 +464,7 @@ class TestCompositionCandidateSelection:
|
||||
coord._search_visible_memories("anything", limit=5)
|
||||
search_mock.assert_called_once()
|
||||
# Second positional arg is the scopes list
|
||||
assert search_mock.call_args.args[1] == [("coordinator", "coord-1")]
|
||||
assert search_mock.call_args.args[1] == [("coordinator", "user-1")]
|
||||
|
||||
|
||||
class TestCompositionRerankFiltersWiring:
|
||||
|
||||
@@ -0,0 +1,214 @@
|
||||
"""Tests for alembic migration 061 (coordinator memories re-keyed to user_id).
|
||||
|
||||
Drives ``command.upgrade`` from a programmatic Alembic config against an
|
||||
isolated SQLite database per test (the 060-test harness pattern), then
|
||||
asserts:
|
||||
|
||||
* a ``scope='coordinator'`` row keyed by a live coordinator's ws_id is
|
||||
re-keyed to that workstream's owner ``user_id``;
|
||||
* rows that cannot be attributed — scope_id matching no workstreams row, or
|
||||
matching one whose ``user_id`` is NULL/empty — are deleted (documented
|
||||
lossy step);
|
||||
* name collisions that would violate ``uq_smem_name_scope`` after the re-key
|
||||
(same name, two coordinator sessions of the same user) keep only the
|
||||
newest ``updated`` row, with ``memory_id`` as the deterministic tiebreak;
|
||||
* the same name under two DIFFERENT users' coordinators survives as two
|
||||
rows (one per user namespace);
|
||||
* non-coordinator scopes are untouched, including a ``user``-scope row whose
|
||||
scope_id equals an owner user_id (the post-rekey value collision the
|
||||
scope column is meant to keep disjoint).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import command
|
||||
from alembic.config import Config
|
||||
|
||||
_MIGRATIONS_DIR = str(
|
||||
Path(__file__).resolve().parent.parent / "turnstone" / "core" / "storage" / "migrations"
|
||||
)
|
||||
|
||||
|
||||
def _alembic_cfg(db_path: Path) -> Config:
|
||||
cfg = Config()
|
||||
cfg.set_main_option("script_location", _MIGRATIONS_DIR)
|
||||
cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}")
|
||||
return cfg
|
||||
|
||||
|
||||
def _seed_ws(conn: sa.Connection, ws_id: str, user_id: object, kind: str = "coordinator") -> None:
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"INSERT INTO workstreams (ws_id, user_id, kind, created, updated) "
|
||||
"VALUES (:ws_id, :user_id, :kind, '2026-06-01T00:00:00', '2026-06-01T00:00:00')"
|
||||
),
|
||||
{"ws_id": ws_id, "user_id": user_id, "kind": kind},
|
||||
)
|
||||
|
||||
|
||||
def _seed_memory(
|
||||
conn: sa.Connection,
|
||||
memory_id: str,
|
||||
name: str,
|
||||
scope: str,
|
||||
scope_id: str,
|
||||
updated: str = "2026-06-01T00:00:00",
|
||||
) -> None:
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"INSERT INTO structured_memories "
|
||||
"(memory_id, name, scope, scope_id, content, created, updated) "
|
||||
"VALUES (:memory_id, :name, :scope, :scope_id, :content, :created, :updated)"
|
||||
),
|
||||
{
|
||||
"memory_id": memory_id,
|
||||
"name": name,
|
||||
"scope": scope,
|
||||
"scope_id": scope_id,
|
||||
"content": f"content of {memory_id}",
|
||||
"created": "2026-06-01T00:00:00",
|
||||
"updated": updated,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _all_memories(engine: sa.Engine) -> dict[str, tuple[str, str, str]]:
|
||||
"""memory_id -> (name, scope, scope_id) for every surviving row."""
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
sa.text("SELECT memory_id, name, scope, scope_id FROM structured_memories")
|
||||
).fetchall()
|
||||
return {r[0]: (r[1], r[2], r[3]) for r in rows}
|
||||
|
||||
|
||||
class TestMigration061:
|
||||
def test_rekeys_to_owner_user_id_and_leaves_other_scopes_alone(self, tmp_path: Path) -> None:
|
||||
db_path = tmp_path / "061-rekey.db"
|
||||
cfg = _alembic_cfg(db_path)
|
||||
command.upgrade(cfg, "060")
|
||||
|
||||
engine = sa.create_engine(f"sqlite:///{db_path}")
|
||||
try:
|
||||
with engine.begin() as conn:
|
||||
_seed_ws(conn, "coord-a", "user-1")
|
||||
_seed_memory(conn, "m1", "runbook", "coordinator", "coord-a")
|
||||
# Other scopes must pass through untouched — including a
|
||||
# user-scope row already keyed by the SAME user_id the
|
||||
# coordinator row is about to be re-keyed onto.
|
||||
_seed_memory(conn, "m2", "runbook", "user", "user-1")
|
||||
_seed_memory(conn, "m3", "runbook", "workstream", "coord-a")
|
||||
_seed_memory(conn, "m4", "runbook", "global", "")
|
||||
|
||||
command.upgrade(cfg, "061")
|
||||
|
||||
mems = _all_memories(engine)
|
||||
assert mems["m1"] == ("runbook", "coordinator", "user-1")
|
||||
assert mems["m2"] == ("runbook", "user", "user-1")
|
||||
assert mems["m3"] == ("runbook", "workstream", "coord-a")
|
||||
assert mems["m4"] == ("runbook", "global", "")
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
def test_deletes_unattributable_rows(self, tmp_path: Path) -> None:
|
||||
db_path = tmp_path / "061-orphans.db"
|
||||
cfg = _alembic_cfg(db_path)
|
||||
command.upgrade(cfg, "060")
|
||||
|
||||
engine = sa.create_engine(f"sqlite:///{db_path}")
|
||||
try:
|
||||
with engine.begin() as conn:
|
||||
_seed_ws(conn, "coord-null", None)
|
||||
_seed_ws(conn, "coord-empty", "")
|
||||
_seed_memory(conn, "m-gone", "a", "coordinator", "no-such-ws")
|
||||
_seed_memory(conn, "m-null", "b", "coordinator", "coord-null")
|
||||
_seed_memory(conn, "m-empty", "c", "coordinator", "coord-empty")
|
||||
|
||||
command.upgrade(cfg, "061")
|
||||
|
||||
assert _all_memories(engine) == {}
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
def test_dedups_same_user_collisions_keeping_newest(self, tmp_path: Path) -> None:
|
||||
db_path = tmp_path / "061-dedup.db"
|
||||
cfg = _alembic_cfg(db_path)
|
||||
command.upgrade(cfg, "060")
|
||||
|
||||
engine = sa.create_engine(f"sqlite:///{db_path}")
|
||||
try:
|
||||
with engine.begin() as conn:
|
||||
_seed_ws(conn, "coord-old", "user-1")
|
||||
_seed_ws(conn, "coord-new", "user-1")
|
||||
_seed_ws(conn, "coord-tie", "user-1")
|
||||
# Same name across two sessions of the same user — only
|
||||
# the newer ``updated`` survives.
|
||||
_seed_memory(
|
||||
conn,
|
||||
"m-old",
|
||||
"plan",
|
||||
"coordinator",
|
||||
"coord-old",
|
||||
updated="2026-06-01T00:00:00",
|
||||
)
|
||||
_seed_memory(
|
||||
conn,
|
||||
"m-new",
|
||||
"plan",
|
||||
"coordinator",
|
||||
"coord-new",
|
||||
updated="2026-06-02T00:00:00",
|
||||
)
|
||||
# Equal ``updated``: memory_id breaks the tie (max wins).
|
||||
_seed_memory(
|
||||
conn,
|
||||
"m-tie-a",
|
||||
"pinned",
|
||||
"coordinator",
|
||||
"coord-new",
|
||||
updated="2026-06-03T00:00:00",
|
||||
)
|
||||
_seed_memory(
|
||||
conn,
|
||||
"m-tie-b",
|
||||
"pinned",
|
||||
"coordinator",
|
||||
"coord-tie",
|
||||
updated="2026-06-03T00:00:00",
|
||||
)
|
||||
# Distinct names never collide — both survive.
|
||||
_seed_memory(conn, "m-keep", "notes", "coordinator", "coord-old")
|
||||
|
||||
command.upgrade(cfg, "061")
|
||||
|
||||
mems = _all_memories(engine)
|
||||
assert "m-old" not in mems
|
||||
assert mems["m-new"] == ("plan", "coordinator", "user-1")
|
||||
assert "m-tie-a" not in mems
|
||||
assert mems["m-tie-b"] == ("pinned", "coordinator", "user-1")
|
||||
assert mems["m-keep"] == ("notes", "coordinator", "user-1")
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
def test_same_name_across_users_survives_as_two_rows(self, tmp_path: Path) -> None:
|
||||
db_path = tmp_path / "061-two-users.db"
|
||||
cfg = _alembic_cfg(db_path)
|
||||
command.upgrade(cfg, "060")
|
||||
|
||||
engine = sa.create_engine(f"sqlite:///{db_path}")
|
||||
try:
|
||||
with engine.begin() as conn:
|
||||
_seed_ws(conn, "coord-u1", "user-1")
|
||||
_seed_ws(conn, "coord-u2", "user-2")
|
||||
_seed_memory(conn, "m-u1", "plan", "coordinator", "coord-u1")
|
||||
_seed_memory(conn, "m-u2", "plan", "coordinator", "coord-u2")
|
||||
|
||||
command.upgrade(cfg, "061")
|
||||
|
||||
mems = _all_memories(engine)
|
||||
assert mems["m-u1"] == ("plan", "coordinator", "user-1")
|
||||
assert mems["m-u2"] == ("plan", "coordinator", "user-2")
|
||||
finally:
|
||||
engine.dispose()
|
||||
+161
-24
@@ -2259,28 +2259,36 @@ class TestCoordinatorMemoryScope:
|
||||
be steered by attackers, so the coord scope must NOT become a delivery
|
||||
channel that injects child-controlled text into the parent's system
|
||||
message.
|
||||
|
||||
The scope is keyed by the coordinator's creator ``user_id`` (NOT its
|
||||
ws_id), so the namespace is durable: every coordinator session the
|
||||
same user runs shares it. The containment gate is the session KIND —
|
||||
children share the parent's user_id and must still be rejected.
|
||||
"""
|
||||
|
||||
def test_coordinator_session_resolves_to_own_ws_id(self, tmp_db):
|
||||
def test_coordinator_session_resolves_to_user_id(self, tmp_db):
|
||||
from turnstone.core.session import ChatSession
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
session = _make_session(
|
||||
ws_id="coord-1",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
assert isinstance(session, ChatSession) # type narrow
|
||||
assert session._resolve_scope_id("coordinator") == "coord-1"
|
||||
assert session._resolve_scope_id("coordinator") == "user-1"
|
||||
|
||||
def test_child_session_resolves_empty(self, tmp_db):
|
||||
"""A child interactive ws of a coord does NOT inherit the
|
||||
coord's scope_id — the row is private to the coord. Children
|
||||
get an empty scope_id which ``_validate_scope`` translates into
|
||||
an explicit reject."""
|
||||
"""A child interactive ws of a coord does NOT inherit the coord
|
||||
scope even though it shares the coord's ``user_id`` — the gate
|
||||
is the session kind, not the scope_id value. Children get an
|
||||
empty scope_id which ``_validate_scope`` translates into an
|
||||
explicit reject."""
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
session = _make_session(
|
||||
ws_id="child-a",
|
||||
user_id="user-1", # same user as the parent coord
|
||||
kind=WorkstreamKind.INTERACTIVE,
|
||||
parent_ws_id="coord-1",
|
||||
)
|
||||
@@ -2288,11 +2296,13 @@ class TestCoordinatorMemoryScope:
|
||||
|
||||
def test_top_level_interactive_resolves_empty(self, tmp_db):
|
||||
"""An IC session with no parent also has no coord context — same
|
||||
empty scope_id, same explicit reject from ``_validate_scope``."""
|
||||
empty scope_id, same explicit reject from ``_validate_scope`` —
|
||||
even when authenticated as a user who owns coordinators."""
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
session = _make_session(
|
||||
ws_id="ws-top",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.INTERACTIVE,
|
||||
parent_ws_id=None,
|
||||
)
|
||||
@@ -2332,6 +2342,7 @@ class TestCoordinatorMemoryScope:
|
||||
|
||||
session = _make_session(
|
||||
ws_id="coord-1",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
assert session._validate_scope("coordinator", "call_1") is None
|
||||
@@ -2339,11 +2350,13 @@ class TestCoordinatorMemoryScope:
|
||||
def test_prepare_memory_save_accepts_coord_scope_for_coord(self, tmp_db):
|
||||
"""The ``save`` action's preparer must round-trip
|
||||
scope='coordinator' through to the execute item with scope_id
|
||||
resolved to the coord's own ws_id."""
|
||||
resolved to the coord's creator user_id (the durable per-user
|
||||
namespace key)."""
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
session = _make_session(
|
||||
ws_id="coord-1",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
item = session._prepare_memory(
|
||||
@@ -2357,16 +2370,19 @@ class TestCoordinatorMemoryScope:
|
||||
)
|
||||
assert "error" not in item
|
||||
assert item["scope"] == "coordinator"
|
||||
assert item["scope_id"] == "coord-1"
|
||||
assert item["scope_id"] == "user-1"
|
||||
|
||||
def test_prepare_memory_save_rejects_coord_scope_for_child(self, tmp_db):
|
||||
"""Children's memory(action='save', scope='coordinator') must
|
||||
return an error item, not silently downgrade to a different
|
||||
scope and not write into the coord's namespace."""
|
||||
scope and not write into the coord's namespace. The child
|
||||
shares the parent's user_id — exactly the credentials a
|
||||
user-keyed scope would accept if kind weren't the gate."""
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
session = _make_session(
|
||||
ws_id="child-a",
|
||||
user_id="user-1", # same user as the parent coord
|
||||
kind=WorkstreamKind.INTERACTIVE,
|
||||
parent_ws_id="coord-1",
|
||||
)
|
||||
@@ -2383,10 +2399,10 @@ class TestCoordinatorMemoryScope:
|
||||
assert "coordinator" in item["error"]
|
||||
|
||||
def test_coord_save_visible_only_to_coord(self, tmp_db):
|
||||
"""A coord-scope memory must be visible to the coord but
|
||||
NOT to its children, NOT to other coords' children, and NOT to
|
||||
unrelated top-level IC sessions. The coord-scope row is
|
||||
private to the coord that owns it."""
|
||||
"""A coord-scope memory must be visible to its user's coordinator
|
||||
sessions (ALL of them — the namespace is per-user durable) but
|
||||
NOT to children (same user!), NOT to unrelated IC sessions, and
|
||||
NOT to another user's coordinators."""
|
||||
from turnstone.core.memory import save_structured_memory
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
@@ -2394,21 +2410,33 @@ class TestCoordinatorMemoryScope:
|
||||
"private_plan",
|
||||
"internal coord notes",
|
||||
scope="coordinator",
|
||||
scope_id="coord-1",
|
||||
scope_id="user-1",
|
||||
)
|
||||
|
||||
coord = _make_session(
|
||||
ws_id="coord-1",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
# The coord sees its own row.
|
||||
# The coord sees its user's row.
|
||||
coord_visible = {m["name"] for m in coord._list_visible_memories()}
|
||||
assert "private_plan" in coord_visible
|
||||
|
||||
# Children of the SAME coord don't see it — closes the
|
||||
# prompt-injection lane.
|
||||
# A LATER coordinator session of the same user (fresh ws_id)
|
||||
# sees the same row — this is the persistence the per-user
|
||||
# keying buys; under ws_id keying this set was always empty.
|
||||
coord_next = _make_session(
|
||||
ws_id="coord-9",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
assert "private_plan" in {m["name"] for m in coord_next._list_visible_memories()}
|
||||
|
||||
# Children of the SAME coord — same user_id — don't see it.
|
||||
# Closes the prompt-injection lane: kind is the gate.
|
||||
child = _make_session(
|
||||
ws_id="child-a",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.INTERACTIVE,
|
||||
parent_ws_id="coord-1",
|
||||
)
|
||||
@@ -2418,15 +2446,17 @@ class TestCoordinatorMemoryScope:
|
||||
# Children of a DIFFERENT coord don't see it (cross-coord).
|
||||
unrelated_child = _make_session(
|
||||
ws_id="child-b",
|
||||
user_id="user-2",
|
||||
kind=WorkstreamKind.INTERACTIVE,
|
||||
parent_ws_id="coord-2",
|
||||
)
|
||||
unrelated_child_visible = {m["name"] for m in unrelated_child._list_visible_memories()}
|
||||
assert "private_plan" not in unrelated_child_visible
|
||||
|
||||
# A different coord doesn't see another coord's row.
|
||||
# Another USER's coordinator doesn't see this user's rows.
|
||||
other_coord = _make_session(
|
||||
ws_id="coord-2",
|
||||
user_id="user-2",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
other_coord_visible = {m["name"] for m in other_coord._list_visible_memories()}
|
||||
@@ -2463,9 +2493,10 @@ class TestCoordinatorMemoryScope:
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
visible = {m["name"] for m in coord._list_visible_memories()}
|
||||
# The coord's own ws_id matching workstream-scope rows must NOT
|
||||
# leak in — coord and IC use different scopes even if their
|
||||
# ids could collide on synthetic test inputs.
|
||||
# ``user_note`` is the sharpest case now: its scope_id
|
||||
# ("user-1") is IDENTICAL to the coord's coordinator scope_id —
|
||||
# the scope COLUMN is what keeps the namespaces disjoint. Same
|
||||
# for ``ws_note`` matching the coord's ws_id.
|
||||
assert "ws_note" not in visible
|
||||
assert "user_note" not in visible
|
||||
assert "global_note" not in visible
|
||||
@@ -2489,7 +2520,7 @@ class TestCoordinatorMemoryScope:
|
||||
"coord_x",
|
||||
"orchestration content",
|
||||
scope="coordinator",
|
||||
scope_id="coord-1",
|
||||
scope_id="user-1",
|
||||
)
|
||||
|
||||
coord = _make_session(
|
||||
@@ -2527,6 +2558,7 @@ class TestCoordinatorMemoryScope:
|
||||
|
||||
coord = _make_session(
|
||||
ws_id="coord-1",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
item = coord._prepare_memory(
|
||||
@@ -2535,7 +2567,7 @@ class TestCoordinatorMemoryScope:
|
||||
)
|
||||
assert "error" not in item
|
||||
assert item["scope"] == "coordinator"
|
||||
assert item["scope_id"] == "coord-1"
|
||||
assert item["scope_id"] == "user-1"
|
||||
|
||||
def test_coord_implicit_walk_only_coordinator(self, tmp_db):
|
||||
"""Coord ``memory(action='get')`` with no explicit scope must
|
||||
@@ -2546,6 +2578,7 @@ class TestCoordinatorMemoryScope:
|
||||
|
||||
coord = _make_session(
|
||||
ws_id="coord-1",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
item = coord._prepare_memory(
|
||||
@@ -2574,6 +2607,110 @@ class TestCoordinatorMemoryScope:
|
||||
scopes = [s for s, _ in item["scopes_to_try"]]
|
||||
assert scopes == ["workstream", "user", "global"]
|
||||
|
||||
def test_coord_memory_persists_across_sessions(self, tmp_db):
|
||||
"""End-to-end through the real save lane: a memory saved by one
|
||||
coordinator session is readable by a LATER coordinator session
|
||||
of the same user (fresh ws_id) — the regression this scope
|
||||
redesign exists to fix. Under ws_id keying the second session
|
||||
was born into an empty namespace every time."""
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
first = _make_session(
|
||||
ws_id="coord-old",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
item = first._prepare_memory(
|
||||
"call_1",
|
||||
{
|
||||
"action": "save",
|
||||
"name": "deploy_runbook",
|
||||
"content": "drain node before rotating certs",
|
||||
"scope": "coordinator",
|
||||
},
|
||||
)
|
||||
assert "error" not in item
|
||||
result = item["execute"](item)
|
||||
assert "Saved" in str(result) or "saved" in str(result).lower()
|
||||
|
||||
# Brand-new coordinator session, new ws_id, same user.
|
||||
second = _make_session(
|
||||
ws_id="coord-new",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
get_item = second._prepare_memory(
|
||||
"call_2",
|
||||
{"action": "get", "name": "deploy_runbook"},
|
||||
)
|
||||
assert "error" not in get_item
|
||||
out = str(get_item["execute"](get_item))
|
||||
assert "drain node before rotating certs" in out
|
||||
|
||||
def test_coordinator_session_requires_user_id(self, tmp_db):
|
||||
"""Anonymous coordinators must not be constructible: the
|
||||
constructor is the host-independent choke point (create,
|
||||
rehydrate, and any future host all pass through it). An empty
|
||||
user_id would otherwise key the durable scope on ``""`` —
|
||||
one namespace shared by every unauthenticated session — and
|
||||
mint child-spawn tokens for a phantom principal."""
|
||||
import pytest
|
||||
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
with pytest.raises(ValueError, match="authenticated user_id"):
|
||||
_make_session(
|
||||
ws_id="coord-anon",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
with pytest.raises(ValueError, match="authenticated user_id"):
|
||||
_make_session(
|
||||
ws_id="coord-anon",
|
||||
user_id="",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
|
||||
def test_validate_scope_backstop_rejects_unauthenticated_coord(self, tmp_db):
|
||||
"""Defense-in-depth behind the constructor guard: if a session
|
||||
ever reaches the memory layer as an unauthenticated coordinator
|
||||
(test double, future host bypass), the save lane is refused at
|
||||
validation and scope resolution stays empty/fail-closed."""
|
||||
from turnstone.core.workstream import WorkstreamKind
|
||||
|
||||
coord = _make_session(
|
||||
ws_id="coord-1",
|
||||
user_id="user-1",
|
||||
kind=WorkstreamKind.COORDINATOR,
|
||||
)
|
||||
coord._user_id = "" # simulate a constructor-bypassing double
|
||||
err = coord._validate_scope("coordinator", "call_1")
|
||||
assert err is not None
|
||||
assert "requires authenticated user identity" in err["error"]
|
||||
assert coord._coordinator_scope_id() == ""
|
||||
item = coord._prepare_memory(
|
||||
"call_1",
|
||||
{"action": "save", "name": "x", "content": "y", "scope": "coordinator"},
|
||||
)
|
||||
assert "error" in item
|
||||
|
||||
# The implicit read lane must fail closed too: the storage
|
||||
# helpers treat a falsy scope_id as "no scope_id filter", so
|
||||
# ("coordinator", "") would otherwise read EVERY user's
|
||||
# coordinator rows. Seed another user's row and prove the
|
||||
# unauthenticated double sees nothing, not everything.
|
||||
from turnstone.core.memory import save_structured_memory
|
||||
|
||||
save_structured_memory(
|
||||
"other_users_row",
|
||||
"must not leak",
|
||||
scope="coordinator",
|
||||
scope_id="user-9",
|
||||
)
|
||||
assert coord._visible_scopes() == []
|
||||
assert coord._visible_memory_count() == 0
|
||||
assert coord._list_visible_memories() == []
|
||||
assert coord._search_visible_memories("leak") == []
|
||||
|
||||
|
||||
class TestMemoryToolAudit:
|
||||
"""Mutating memory tool actions emit audit rows.
|
||||
|
||||
@@ -361,6 +361,7 @@ def test_chatsession_coordinator_kind_excludes_interactive_tools(tmp_db):
|
||||
temperature=0.5,
|
||||
max_tokens=4096,
|
||||
tool_timeout=30,
|
||||
user_id="user-1", # the constructor refuses anonymous coordinators
|
||||
kind="coordinator",
|
||||
)
|
||||
names = {t["function"]["name"] for t in sess._tools}
|
||||
@@ -401,6 +402,7 @@ def test_chatsession_coordinator_kind_does_not_merge_mcp_tools(tmp_db):
|
||||
temperature=0.5,
|
||||
max_tokens=4096,
|
||||
tool_timeout=30,
|
||||
user_id="user-1", # the constructor refuses anonymous coordinators
|
||||
kind="coordinator",
|
||||
mcp_client=mcp_client,
|
||||
)
|
||||
|
||||
@@ -4507,9 +4507,16 @@ def _bootstrap_coord_subsystem(
|
||||
return ConsoleCoordinatorUI(ws_id=ws.id, user_id=ws.user_id or "")
|
||||
|
||||
def _coord_client_factory(ws_id: str, user_id: str) -> CoordinatorClient:
|
||||
# No empty-uid fallback: CoordinatorTokenManager's contract is
|
||||
# "sub — the coordinator's real creator user_id", and
|
||||
# ChatSession.__init__ refuses to construct an anonymous
|
||||
# coordinator before any token is minted (minting is lazy).
|
||||
# Masking an empty uid as a phantom principal here would hand
|
||||
# a read/write/approve + admin.coordinator token to a user
|
||||
# that doesn't exist and persist its children under that name.
|
||||
ttl = int(config_store.get("coordinator.session_jwt_ttl_seconds"))
|
||||
tm = CoordinatorTokenManager(
|
||||
user_id=user_id or "system",
|
||||
user_id=user_id,
|
||||
scopes=frozenset({"read", "write", "approve"}),
|
||||
permissions=frozenset({"admin.coordinator"}),
|
||||
secret=jwt_secret,
|
||||
|
||||
+78
-12
@@ -945,6 +945,20 @@ class ChatSession:
|
||||
parent_ws_id: str | None = None,
|
||||
coord_client: Any = None,
|
||||
):
|
||||
if kind == WorkstreamKind.COORDINATOR and not user_id:
|
||||
# Coordinators carry real authority — they mint child-spawn
|
||||
# tokens as their creator and own a durable per-user memory
|
||||
# namespace — so an anonymous one must never exist. The
|
||||
# empty-string user_id sentinel is an interactive-only lane
|
||||
# (CLI / eval / placeholder sessions); every coordinator
|
||||
# host authenticates (console HTTP create 401s an empty
|
||||
# principal), leaving rehydration of a corrupt or legacy
|
||||
# row as the path this guard surfaces.
|
||||
raise ValueError(
|
||||
"coordinator sessions require an authenticated user_id; "
|
||||
f"refusing to construct an anonymous coordinator (ws_id={ws_id!r}). "
|
||||
"If this is a persisted legacy row, delete or close it."
|
||||
)
|
||||
self.client = client
|
||||
self.model = model
|
||||
# Coordinator plumbing: populated by the console's session factory
|
||||
@@ -7281,6 +7295,16 @@ class ChatSession:
|
||||
content (MCP tool output, attachments) which can be steered to
|
||||
plant instructions, and the new scope must not become a
|
||||
delivery channel back into the parent's prompt.
|
||||
|
||||
The containment gate is the session KIND (children are always
|
||||
INTERACTIVE \u2014 :meth:`_validate_scope` rejects them before this
|
||||
resolver runs), not secrecy of the scope_id value. That is
|
||||
what lets the coord scope key on the durable ``user_id``
|
||||
(shared with children, visible cluster-wide as display
|
||||
metadata) without widening the write surface: no lane \u2014 memory
|
||||
tool or REST (``_VALID_MEMORY_SCOPES`` in ``server.py`` omits
|
||||
``coordinator``) \u2014 accepts a caller-supplied coordinator
|
||||
scope_id.
|
||||
"""
|
||||
if scope == "workstream":
|
||||
return self._ws_id
|
||||
@@ -7291,20 +7315,27 @@ class ChatSession:
|
||||
return ""
|
||||
|
||||
def _coordinator_scope_id(self) -> str:
|
||||
"""Return the ws_id anchoring the ``coordinator`` memory scope, or ``""``.
|
||||
"""Return the user_id anchoring the ``coordinator`` memory scope, or ``""``.
|
||||
|
||||
Only a coordinator session has a coord scope \u2014 returns
|
||||
``self._ws_id`` for ``kind == COORDINATOR``, ``""`` otherwise.
|
||||
Children of a coord get an empty scope_id, which
|
||||
:meth:`_validate_scope` translates into an explicit reject \u2014
|
||||
children must use ``workstream`` or ``user`` scope for their
|
||||
own memories.
|
||||
``self._user_id`` for ``kind == COORDINATOR``, ``""`` otherwise.
|
||||
Keying on the user (not the ws_id) makes the namespace durable:
|
||||
every coordinator session the same user runs shares one
|
||||
orchestration memory, so notes survive close/reopen. Children
|
||||
of a coord get an empty scope_id, which :meth:`_validate_scope`
|
||||
translates into an explicit reject \u2014 children must use
|
||||
``workstream`` or ``user`` scope for their own memories.
|
||||
|
||||
``""`` for an unauthenticated coordinator is unreachable in
|
||||
practice (``__init__`` refuses to construct one) but kept
|
||||
fail-closed: an empty scope_id never resolves to a readable or
|
||||
writable namespace.
|
||||
|
||||
See :meth:`_resolve_scope_id`'s docstring for the security
|
||||
rationale (cross-session prompt-injection containment).
|
||||
"""
|
||||
if self._kind == WorkstreamKind.COORDINATOR:
|
||||
return self._ws_id
|
||||
return self._user_id
|
||||
return ""
|
||||
|
||||
def _validate_scope(self, scope: str, call_id: str) -> dict[str, Any] | None:
|
||||
@@ -7316,9 +7347,11 @@ class ChatSession:
|
||||
doesn't accidentally mutate or read user-context rows.
|
||||
|
||||
Interactive sessions reject ``coordinator`` for the symmetric
|
||||
reason \u2014 coord-scope rows are private to a coordinator
|
||||
session, and an IC writer could otherwise be a cross-session
|
||||
prompt-injection lane into the parent coord's system message.
|
||||
reason \u2014 coord-scope rows belong to a per-user namespace read
|
||||
only by that user's COORDINATOR sessions, and an IC writer
|
||||
(children share the parent's user_id, so the kind check is the
|
||||
gate) could otherwise be a cross-session prompt-injection lane
|
||||
into the parent coord's system message.
|
||||
"""
|
||||
if scope == "user" and not self._user_id:
|
||||
return {
|
||||
@@ -7329,6 +7362,27 @@ class ChatSession:
|
||||
"needs_approval": False,
|
||||
"error": "Error: 'user' scope requires authenticated user identity",
|
||||
}
|
||||
if (
|
||||
scope == "coordinator"
|
||||
and self._kind == WorkstreamKind.COORDINATOR
|
||||
and not self._user_id
|
||||
):
|
||||
# Backstop for the save lane: search/list reject empty
|
||||
# scope_ids in _exec_memory, but save would otherwise write
|
||||
# a ("coordinator", "") row shared by every unauthenticated
|
||||
# session. Unreachable through real hosts (__init__ refuses
|
||||
# COORDINATOR without a user_id); guards test doubles and
|
||||
# future hosts. Kind-scoped so non-coordinator callers keep
|
||||
# the clearer kind-mismatch error below regardless of their
|
||||
# auth state.
|
||||
return {
|
||||
"call_id": call_id,
|
||||
"func_name": "memory",
|
||||
"header": "\u2717 memory: coordinator scope requires authentication",
|
||||
"preview": "",
|
||||
"needs_approval": False,
|
||||
"error": "Error: 'coordinator' scope requires authenticated user identity",
|
||||
}
|
||||
if self._kind == WorkstreamKind.COORDINATOR and scope != "coordinator":
|
||||
return {
|
||||
"call_id": call_id,
|
||||
@@ -7397,7 +7451,10 @@ class ChatSession:
|
||||
message, which the coord shouldn't be reasoning over.
|
||||
"""
|
||||
if self._kind == WorkstreamKind.COORDINATOR:
|
||||
return count_structured_memories(scope="coordinator", scope_id=self._ws_id)
|
||||
scope_id = self._coordinator_scope_id()
|
||||
if not scope_id:
|
||||
return 0
|
||||
return count_structured_memories(scope="coordinator", scope_id=scope_id)
|
||||
n = count_structured_memories(scope="global")
|
||||
n += count_structured_memories(scope="workstream", scope_id=self._ws_id)
|
||||
if self._user_id:
|
||||
@@ -7412,7 +7469,16 @@ class ChatSession:
|
||||
the single-query visibility helpers.
|
||||
"""
|
||||
if self._kind == WorkstreamKind.COORDINATOR:
|
||||
return [("coordinator", self._ws_id)]
|
||||
scope_id = self._coordinator_scope_id()
|
||||
# Fail-closed on an empty scope_id (unreachable through real
|
||||
# hosts — __init__ refuses anonymous coordinators): the
|
||||
# storage helpers treat a falsy scope_id as "no scope_id
|
||||
# filter" (that's how ``global`` works), so passing
|
||||
# ("coordinator", "") through would read EVERY user's
|
||||
# coordinator rows instead of none.
|
||||
if not scope_id:
|
||||
return []
|
||||
return [("coordinator", scope_id)]
|
||||
scopes: list[tuple[str, str]] = [("global", ""), ("workstream", self._ws_id)]
|
||||
if self._user_id:
|
||||
scopes.append(("user", self._user_id))
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
"""Re-key ``coordinator``-scope memories from ws_id to the owner's user_id.
|
||||
|
||||
The ``coordinator`` memory scope used to anchor on the coordinator session's
|
||||
``ws_id``, so the namespace was born empty with every new coordinator session
|
||||
and its rows were orphaned the moment that session closed — coordinator
|
||||
memory never actually persisted. The scope now anchors on the coordinator's
|
||||
creator ``user_id`` (one durable orchestration namespace per user, shared by
|
||||
all of that user's coordinator sessions). This migration carries the
|
||||
existing rows across:
|
||||
|
||||
1. **Delete unattributable rows** — ``scope='coordinator'`` rows whose
|
||||
``scope_id`` matches no ``workstreams.ws_id``, or whose owning workstream
|
||||
has a NULL/empty ``user_id`` (a pre-auth-guard anomaly: coordinator
|
||||
sessions now refuse to construct without an authenticated user). Such
|
||||
rows cannot be assigned to any user and would be unreachable forever
|
||||
under user keying — for a private, fail-closed scope they are deleted
|
||||
rather than left as permanent dead rows. This is intentionally lossy.
|
||||
2. **Dedup colliding names** — two coordinator sessions of the same user
|
||||
could each hold a memory with the same ``name`` (distinct ws_id
|
||||
scope_ids). After re-keying both would map to the same
|
||||
``(name, 'coordinator', user_id)`` identity and violate
|
||||
``uq_smem_name_scope``; keep the most recently ``updated`` row (ties
|
||||
broken by ``memory_id`` for determinism) and delete the rest.
|
||||
``updated`` is a fixed-width ``%Y-%m-%dT%H:%M:%S`` string, so
|
||||
lexicographic order is chronological.
|
||||
3. **Re-key** — set ``scope_id`` to the owning workstream's ``user_id``.
|
||||
|
||||
``downgrade()`` is a documented no-op: there is no schema delta, and the
|
||||
data transform is not reversible (deleted orphans are gone; deduped rows are
|
||||
gone; the many-old-namespaces → one-user-namespace collapse cannot be
|
||||
unsplit). Pre-061 code simply sees the user-keyed rows as not-visible, the
|
||||
same way it saw any closed session's rows.
|
||||
|
||||
Revision ID: 061
|
||||
Revises: 060
|
||||
Create Date: 2026-06-12
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision = "061"
|
||||
down_revision = "060"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
|
||||
# 1. Drop rows that cannot be attributed to a user.
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"DELETE FROM structured_memories "
|
||||
"WHERE scope = 'coordinator' "
|
||||
"AND NOT EXISTS ("
|
||||
" SELECT 1 FROM workstreams w "
|
||||
" WHERE w.ws_id = structured_memories.scope_id "
|
||||
" AND w.user_id IS NOT NULL AND w.user_id != ''"
|
||||
")"
|
||||
)
|
||||
)
|
||||
|
||||
# 2. Within each post-rekey identity (name, owner user_id), keep only
|
||||
# the newest row. Every remaining row joins to an owning
|
||||
# workstream with a non-empty user_id (step 1 guarantees it).
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"DELETE FROM structured_memories "
|
||||
"WHERE scope = 'coordinator' "
|
||||
"AND EXISTS ("
|
||||
" SELECT 1 "
|
||||
" FROM structured_memories s2 "
|
||||
" JOIN workstreams w2 ON w2.ws_id = s2.scope_id "
|
||||
" JOIN workstreams w1 ON w1.ws_id = structured_memories.scope_id "
|
||||
" WHERE s2.scope = 'coordinator' "
|
||||
" AND s2.name = structured_memories.name "
|
||||
" AND w2.user_id = w1.user_id "
|
||||
" AND s2.memory_id != structured_memories.memory_id "
|
||||
" AND (s2.updated > structured_memories.updated "
|
||||
" OR (s2.updated = structured_memories.updated "
|
||||
" AND s2.memory_id > structured_memories.memory_id))"
|
||||
")"
|
||||
)
|
||||
)
|
||||
|
||||
# 3. Re-key the survivors onto their owner's user_id.
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"UPDATE structured_memories "
|
||||
"SET scope_id = ("
|
||||
" SELECT w.user_id FROM workstreams w "
|
||||
" WHERE w.ws_id = structured_memories.scope_id"
|
||||
") "
|
||||
"WHERE scope = 'coordinator'"
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
# Irreversible data transform — see module docstring. No schema delta,
|
||||
# so there is nothing structural to undo either.
|
||||
pass
|
||||
@@ -55,11 +55,11 @@
|
||||
}
|
||||
},
|
||||
"coordinator": {
|
||||
"description": "Persistent orchestration memory for this coordinator session. Actions: 'save' stores a memory, 'get' retrieves full content by name, 'search' finds memories by query, 'delete' removes a memory, 'list' shows all memories. Use 'get' to read full content — search/list truncate previews to 200 chars. Memories have a type (user/project/feedback/reference). Coordinator memories are private to this coordinator and survive across its turns; they are NOT visible to its child workstreams.",
|
||||
"description": "Persistent orchestration memory shared by all of your user's coordinator sessions. Actions: 'save' stores a memory, 'get' retrieves full content by name, 'search' finds memories by query, 'delete' removes a memory, 'list' shows all memories. Use 'get' to read full content — search/list truncate previews to 200 chars. Memories have a type (user/project/feedback/reference). Coordinator memories survive across coordinator sessions — save orchestration knowledge worth keeping (recurring procedures, environment facts, lessons from past runs). They are NOT visible to child workstreams.",
|
||||
"parameter_overrides": {
|
||||
"scope": {
|
||||
"enum": ["coordinator"],
|
||||
"description": "Always 'coordinator' for coord sessions — coord memories are isolated to the coordinator's own orchestration namespace. This field can be omitted; it defaults to 'coordinator'."
|
||||
"description": "Always 'coordinator' for coord sessions — the per-user orchestration namespace, durable across coordinator sessions. This field can be omitted; it defaults to 'coordinator'."
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user