fix(mcp): make get_note / get_task share-aware
CI & Build / Python lint (push) Successful in 3s
CI & Build / integration (push) Successful in 31s
CI & Build / TypeScript typecheck (push) Successful in 42s
CI & Build / Python tests (push) Successful in 56s
CI & Build / Build & push image (push) Successful in 1m0s
CI & Build / Python lint (push) Successful in 3s
CI & Build / integration (push) Successful in 31s
CI & Build / TypeScript typecheck (push) Successful in 42s
CI & Build / Python tests (push) Successful in 56s
CI & Build / Build & push image (push) Successful in 1m0s
The injected menu tells the agent to open any hit with get_note(id), and that menu can list a collaborator's record reached through a shared project. The fetch was still owner-only, so those lines answered "not found" — for a record the same user opens fine in the browser. The agent path was strictly narrower than the web path for the same id. Same boundary miss as #2093: the list side was widened for sharing, the fetch side wasn't. Both tools now resolve through get_note_for_user, apply the trash filter themselves (permission resolution says nothing about liveness), and attach describe_provenance so a shared record arrives marked as someone else's rather than passing as the caller's. routes/tasks.py's parent-title lookup had the same narrowness: a shared subtask rendered as an orphan when its parent was equally shared. Four fetch tests across three files were patching notes_svc.get_note and had to be retargeted — note 2109's third sub-case, caught by grepping tests/ for the old name before pushing rather than by CI (#2159). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RLwAaV4DQEmVyn496HnEvt
This commit is contained in:
@@ -17,11 +17,18 @@ def _bind_user():
|
||||
_user_id_ctx.reset(token)
|
||||
|
||||
|
||||
def _fake_note(**overrides) -> MagicMock:
|
||||
def _fake_note(*, user_id: int = 7, **overrides) -> MagicMock:
|
||||
note = MagicMock()
|
||||
base = {"id": 1, "title": "t", "body": "b", "tags": [], "is_task": False}
|
||||
base.update(overrides)
|
||||
note.to_dict.return_value = base
|
||||
# Real values, not auto-attributes: get_note reads deleted_at, and compares
|
||||
# user_id against the bound caller to decide whether to attach a shared/owner
|
||||
# marker. A MagicMock is truthy on both, so it would read as another user's
|
||||
# trashed note and reach for the DB (note 2109).
|
||||
note.id = base["id"]
|
||||
note.user_id = user_id
|
||||
note.deleted_at = None
|
||||
return note
|
||||
|
||||
|
||||
@@ -108,24 +115,61 @@ async def test_list_notes_limit_clamped():
|
||||
async def test_get_note_returns_dict():
|
||||
fake = _fake_note(id=5, title="found")
|
||||
with patch(
|
||||
"scribe.mcp.tools.notes.notes_svc.get_note",
|
||||
AsyncMock(return_value=fake),
|
||||
"scribe.mcp.tools.notes.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=(fake, "owner")),
|
||||
):
|
||||
out = await get_note(note_id=5)
|
||||
assert out["id"] == 5
|
||||
assert out["title"] == "found"
|
||||
# Own record: no provenance noise.
|
||||
assert "shared" not in out
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_note_opens_a_shared_record_and_says_whose_it_is():
|
||||
"""The injected menu tells the agent to open any hit with get_note(id), and
|
||||
that menu can list a collaborator's note reached through a shared project.
|
||||
An owner-only fetch would answer "not found" for a record the agent was just
|
||||
handed — and the web UI opens the same note fine."""
|
||||
theirs = _fake_note(id=5, title="Their note", user_id=9)
|
||||
with patch(
|
||||
"scribe.mcp.tools.notes.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=(theirs, "viewer")),
|
||||
), patch(
|
||||
"scribe.services.access.describe_provenance",
|
||||
AsyncMock(return_value={"shared": True, "owner": "alex",
|
||||
"permission": "viewer"}),
|
||||
):
|
||||
out = await get_note(note_id=5)
|
||||
assert out["shared"] is True
|
||||
assert out["owner"] == "alex"
|
||||
assert out["permission"] == "viewer"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_note_raises_when_not_found():
|
||||
with patch(
|
||||
"scribe.mcp.tools.notes.notes_svc.get_note",
|
||||
"scribe.mcp.tools.notes.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=None),
|
||||
):
|
||||
with pytest.raises(ValueError, match="note 999 not found"):
|
||||
await get_note(note_id=999)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_note_treats_a_trashed_note_as_missing():
|
||||
"""get_note_for_user resolves permission, not liveness — the trash filter is
|
||||
the caller's to apply."""
|
||||
trashed = _fake_note(id=5)
|
||||
trashed.deleted_at = "2026-07-01T00:00:00Z"
|
||||
with patch(
|
||||
"scribe.mcp.tools.notes.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=(trashed, "owner")),
|
||||
):
|
||||
with pytest.raises(ValueError, match="note 5 not found"):
|
||||
await get_note(note_id=5)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_note_passes_through():
|
||||
fake = _fake_note(id=10, title="new")
|
||||
|
||||
@@ -24,16 +24,25 @@ async def test_start_planning_tool_delegates_to_service():
|
||||
assert mock.call_args.kwargs == {"user_id": 7, "project_id": 3, "title": "Plan it"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_task_augments_plan_with_rules():
|
||||
def _plan_note(task_kind: str):
|
||||
note = MagicMock()
|
||||
note.parent_id = None
|
||||
note.project_id = 3
|
||||
note.to_dict.return_value = {"id": 9, "task_kind": "plan", "project_id": 3}
|
||||
note.id = 9
|
||||
# Real values — get_task reads deleted_at and compares user_id to the bound
|
||||
# caller, and a MagicMock is truthy on both (note 2109).
|
||||
note.user_id = 7
|
||||
note.deleted_at = None
|
||||
note.to_dict.return_value = {"id": 9, "task_kind": task_kind, "project_id": 3}
|
||||
return note
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_task_augments_plan_with_rules():
|
||||
applicable = {"rules": [{"id": 1, "title": "r"}], "truncated": False,
|
||||
"subscribed_rulebooks": [{"id": 2, "title": "rb"}]}
|
||||
with patch("scribe.mcp.tools.tasks.notes_svc.get_note",
|
||||
AsyncMock(return_value=note)), \
|
||||
with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=(_plan_note("plan"), "owner"))), \
|
||||
patch("scribe.mcp.tools.tasks.rulebooks_svc.get_applicable_rules",
|
||||
AsyncMock(return_value=applicable)):
|
||||
from scribe.mcp.tools.tasks import get_task
|
||||
@@ -45,12 +54,8 @@ async def test_get_task_augments_plan_with_rules():
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_task_work_kind_has_no_rules():
|
||||
note = MagicMock()
|
||||
note.parent_id = None
|
||||
note.project_id = 3
|
||||
note.to_dict.return_value = {"id": 9, "task_kind": "work", "project_id": 3}
|
||||
with patch("scribe.mcp.tools.tasks.notes_svc.get_note",
|
||||
AsyncMock(return_value=note)), \
|
||||
with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=(_plan_note("work"), "owner"))), \
|
||||
patch("scribe.mcp.tools.tasks.rulebooks_svc.get_applicable_rules",
|
||||
AsyncMock()) as mock_rules:
|
||||
from scribe.mcp.tools.tasks import get_task
|
||||
|
||||
@@ -18,7 +18,8 @@ def _bind_user():
|
||||
_user_id_ctx.reset(token)
|
||||
|
||||
|
||||
def _fake_task(*, parent_id: int | None = None, **overrides) -> MagicMock:
|
||||
def _fake_task(*, parent_id: int | None = None, user_id: int = 7,
|
||||
**overrides) -> MagicMock:
|
||||
n = MagicMock()
|
||||
n.parent_id = parent_id
|
||||
base = {
|
||||
@@ -29,6 +30,12 @@ def _fake_task(*, parent_id: int | None = None, **overrides) -> MagicMock:
|
||||
base.update(overrides)
|
||||
n.to_dict.return_value = base
|
||||
n.title = base["title"]
|
||||
n.id = base["id"]
|
||||
# Real values, not auto-attributes: get_task reads deleted_at and compares
|
||||
# user_id against the bound caller for the shared/owner marker — a MagicMock
|
||||
# is truthy on both (note 2109).
|
||||
n.user_id = user_id
|
||||
n.deleted_at = None
|
||||
return n
|
||||
|
||||
|
||||
@@ -63,12 +70,13 @@ async def test_list_tasks_empty_status_means_no_filter():
|
||||
async def test_get_task_with_no_parent_returns_null_parent_title():
|
||||
fake = _fake_task(id=5, title="solo", parent_id=None)
|
||||
with patch(
|
||||
"scribe.mcp.tools.tasks.notes_svc.get_note",
|
||||
AsyncMock(return_value=fake),
|
||||
"scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=(fake, "owner")),
|
||||
):
|
||||
out = await get_task(task_id=5)
|
||||
assert out["parent_title"] is None
|
||||
assert out["title"] == "solo"
|
||||
assert "shared" not in out
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -76,9 +84,9 @@ async def test_get_task_enriches_with_parent_title():
|
||||
"""When parent_id is set, get_task fetches the parent and adds parent_title."""
|
||||
child = _fake_task(id=10, title="child", parent_id=5)
|
||||
parent = _fake_task(id=5, title="parent of 10", parent_id=None)
|
||||
# get_note is called twice: once for child, once for parent
|
||||
mock_get = AsyncMock(side_effect=[child, parent])
|
||||
with patch("scribe.mcp.tools.tasks.notes_svc.get_note", mock_get):
|
||||
# fetched twice: once for the child, once for the parent
|
||||
mock_get = AsyncMock(side_effect=[(child, "owner"), (parent, "owner")])
|
||||
with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user", mock_get):
|
||||
out = await get_task(task_id=10)
|
||||
assert out["parent_title"] == "parent of 10"
|
||||
assert mock_get.await_count == 2
|
||||
@@ -88,16 +96,34 @@ async def test_get_task_enriches_with_parent_title():
|
||||
async def test_get_task_parent_missing_returns_null():
|
||||
"""If parent_id is set but the parent is gone (orphaned), parent_title is None."""
|
||||
child = _fake_task(id=10, parent_id=5)
|
||||
mock_get = AsyncMock(side_effect=[child, None])
|
||||
with patch("scribe.mcp.tools.tasks.notes_svc.get_note", mock_get):
|
||||
mock_get = AsyncMock(side_effect=[(child, "owner"), None])
|
||||
with patch("scribe.mcp.tools.tasks.notes_svc.get_note_for_user", mock_get):
|
||||
out = await get_task(task_id=10)
|
||||
assert out["parent_title"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_task_opens_a_shared_task_and_says_whose_it_is():
|
||||
"""A task in a shared project opens in the web UI, so the agent path must not
|
||||
answer "not found" for the same id — and must say it isn't the caller's."""
|
||||
theirs = _fake_task(id=5, title="Their task", user_id=9)
|
||||
with patch(
|
||||
"scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=(theirs, "viewer")),
|
||||
), patch(
|
||||
"scribe.services.access.describe_provenance",
|
||||
AsyncMock(return_value={"shared": True, "owner": "alex",
|
||||
"permission": "viewer"}),
|
||||
):
|
||||
out = await get_task(task_id=5)
|
||||
assert out["shared"] is True
|
||||
assert out["owner"] == "alex"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_task_raises_when_not_found():
|
||||
with patch(
|
||||
"scribe.mcp.tools.tasks.notes_svc.get_note",
|
||||
"scribe.mcp.tools.tasks.notes_svc.get_note_for_user",
|
||||
AsyncMock(return_value=None),
|
||||
):
|
||||
with pytest.raises(ValueError, match="task 999 not found"):
|
||||
|
||||
Reference in New Issue
Block a user