feat(6854): the server says which project a pending act would change, and whether this session may do it

GET /api/plugin/write-scope answers the plugin's fence. The session's project
comes from repo / project_id, as every plugin endpoint reads it. The act is
either:
- a directory (target_repo / target_project_id), for a file edit or a git
  command; or
- an MCP call (tool, plus input: the call's short arguments).

It returns allowed, the other projects the act would change, and the reason
the operator is asked. It is a GET so a read-scoped key still fences file
edits and git.

services/write_scope.WRITE_TARGETS lists every write tool with the arguments
that locate its write: a project id, or a task, note, milestone, System, rule,
relation, trash batch, repo or family decision, each traced to its project. A
write that lands in no project (install settings, user-level records, the
judgments a session is asked for anywhere) has an empty entry. The one
cross-project write allowed is create_task(kind="issue") as a new, unplanned,
open issue. A project the caller cannot read is never named.

tests/test_write_scope.py:
- requires an entry for every write tool;
- requires every declared argument to exist on the tool;
- refuses a locating argument (project_id, task_id, milestone_id and the
  rest) that a write tool takes without declaring it.

The integration test traces each locator from real rows.

Milestone 583, step 2.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_integration_write_scope.py

@@ -0,0 +1,86 @@
+"""Real-Postgres proof that a write is traced to the project it lands in.
+
+Two projects of one user: the session's own and another. Each locator kind is
+resolved from a real row, and check_act answers for a session standing in
+the first.
+"""
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from grimoire.models import async_session
+from grimoire.models.milestone import Milestone
+from grimoire.models.note import Note
+from grimoire.models.project import Project
+from grimoire.models.system import System
+from grimoire.services import write_scope
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+OWNER = "write_scope_owner"
+BATCH = "write-scope-test-batch"
+
+
+@pytest_asyncio.fixture
+async def two_projects():
+    async with async_session() as s:
+        user = await ensure_user(s, OWNER)
+        for model in (Note, Milestone, System, Project):
+            for row in (await s.execute(select(model).where(model.user_id == user.id))).scalars():
+                await s.delete(row)
+        await s.flush()
+        home = Project(user_id=user.id, title="Home")
+        other = Project(user_id=user.id, title="Elsewhere")
+        s.add_all([home, other])
+        await s.flush()
+        plan = Milestone(user_id=user.id, project_id=other.id, title="Their plan")
+        area = System(user_id=user.id, project_id=other.id, name="Their area")
+        s.add_all([plan, area])
+        await s.flush()
+        theirs = Note(user_id=user.id, project_id=other.id, milestone_id=plan.id,
+                      title="Their task", body="", note_type="note", task_kind="work",
+                      status="todo")
+        mine = Note(user_id=user.id, project_id=home.id, title="My task", body="",
+                    note_type="note", task_kind="work", status="todo")
+        trashed = Note(user_id=user.id, project_id=other.id, title="Trashed", body="",
+                       note_type="note", deleted_batch_id=BATCH)
+        s.add_all([theirs, mine, trashed])
+        await s.commit()
+        return {"user": user.id, "home": home.id, "other": other.id, "plan": plan.id,
+                "area": area.id, "theirs": theirs.id, "mine": mine.id}
+
+
+async def test_each_locator_finds_the_project_of_the_row_it_names(two_projects):
+    ids = two_projects
+    other, home, uid = ids["other"], ids["home"], ids["user"]
+
+    async def lands(tool, **args):
+        return await write_scope.tool_targets(uid, tool, args)
+
+    assert await lands("add_task_log", task_id=ids["theirs"]) == {other}
+    assert await lands("update_task", task_id=ids["mine"]) == {home}
+    assert await lands("update_task", task_id=ids["mine"], milestone_id=ids["plan"]) == {home, other}
+    assert await lands("delete_milestone", milestone_id=ids["plan"]) == {other}
+    assert await lands("update_system", system_id=ids["area"]) == {other}
+    assert await lands("restore", deleted_batch_id=BATCH) == {other}
+    assert await lands("create_rule", topic_id=1) == set()
+
+
+async def test_a_session_in_one_project_is_asked_before_writing_in_the_other(two_projects):
+    ids = two_projects
+    uid, home, other = ids["user"], ids["home"], ids["other"]
+
+    own = await write_scope.check_act(uid, home, tool="add_task_log",
+                                      args={"task_id": ids["mine"]})
+    assert own["allowed"] is True
+
+    theirs = await write_scope.check_act(uid, home, tool="add_task_log",
+                                         args={"task_id": ids["theirs"]})
+    assert theirs["allowed"] is False
+    assert theirs["targets"] == [{"id": other, "title": "Elsewhere"}]
+    assert "Home" in theirs["reason"] and "Elsewhere" in theirs["reason"]
+
+    issue = await write_scope.check_act(uid, home, tool="create_task",
+                                        args={"project_id": other, "kind": "issue"})
+    assert issue["allowed"] is True and issue["issue"] is True

tests/test_write_scope.py

@@ -0,0 +1,207 @@
+"""Where a write lands, so the plugin can hold a session to its own project.
+
+The guards here keep services/write_scope.WRITE_TARGETS whole: every write
+tool says which of its arguments locate the write, and an argument that would
+locate one can't be left out. The rest covers the one cross-project write
+allowed (filing an issue) and the answer check_act gives.
+"""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from quart import g
+
+from grimoire.services import write_scope
+from grimoire.services.write_scope import (
+    LOCATOR_NAMES, LOCATORS, WRITE_TARGETS, check_act, is_issue_filing,
+)
+
+
+def _write_tool_params() -> dict[str, set[str]]:
+    from grimoire.mcp.server import _WRITE_TOOLS, build_mcp_server
+
+    manager = build_mcp_server()._tool_manager
+    return {
+        name: set(manager.get_tool(name).parameters.get("properties", {}))
+        for name in _WRITE_TOOLS
+    }
+
+
+def test_every_write_tool_says_where_it_writes():
+    from grimoire.mcp.server import _WRITE_TOOLS
+
+    missing = sorted(_WRITE_TOOLS - set(WRITE_TARGETS))
+    assert not missing, (
+        f"write tools with no entry in WRITE_TARGETS: {missing}. Say which "
+        "arguments locate each one's write, or {} when it lands in no project."
+    )
+    extra = sorted(set(WRITE_TARGETS) - _WRITE_TOOLS)
+    assert not extra, f"WRITE_TARGETS names tools that aren't write tools: {extra}"
+
+
+def test_every_declared_argument_is_one_the_tool_takes_and_has_a_known_kind():
+    params = _write_tool_params()
+    for tool, locators in WRITE_TARGETS.items():
+        for arg, kind in locators.items():
+            assert arg in params[tool], f"{tool} has no argument {arg!r}"
+            assert kind in LOCATORS, f"{tool}.{arg}: unknown locator {kind!r}"
+
+
+def test_an_argument_that_would_locate_a_write_cannot_be_left_out():
+    params = _write_tool_params()
+    undeclared = sorted(
+        f"{tool}.{arg}"
+        for tool, names in params.items()
+        for arg in names & LOCATOR_NAMES
+        if arg not in WRITE_TARGETS[tool]
+    )
+    assert not undeclared, (
+        f"write tools take these locating arguments without declaring them in "
+        f"WRITE_TARGETS: {undeclared}"
+    )
+
+
+def test_the_locator_scan_sees_the_arguments_it_guards():
+    # The guard above passes vacuously if the scan finds no parameters at all.
+    params = _write_tool_params()
+    assert "task_id" in params["update_task"] & LOCATOR_NAMES
+    assert "project_id" in params["create_task"] & LOCATOR_NAMES
+
+
+@pytest.mark.parametrize("args, expected", [
+    ({"project_id": 7, "kind": "issue"}, True),
+    ({"project_id": 7, "kind": "issue", "status": "todo"}, True),
+    ({"project_id": 7, "kind": "work"}, False),
+    ({"project_id": 7}, False),
+    ({"project_id": 7, "kind": "issue", "status": "in_progress"}, False),
+    ({"project_id": 7, "kind": "issue", "milestone_id": 3}, False),
+    ({"project_id": 7, "kind": "issue", "parent_id": 9}, False),
+    ({"project_id": 7, "kind": "issue", "milestone_id": 0}, True),
+])
+def test_filing_an_issue_is_a_new_open_unplanned_issue(args, expected):
+    assert is_issue_filing("create_task", args) is expected
+
+
+def test_only_create_task_files_an_issue():
+    assert is_issue_filing("update_task", {"task_id": 4, "kind": "issue"}) is False
+    assert is_issue_filing("create_records", {"project_id": 7, "kind": "issue"}) is False
+
+
+async def test_ids_of_zero_or_below_locate_nothing():
+    # update_task's project_id=0 leaves it unchanged and -1 clears it.
+    assert await write_scope.tool_targets(1, "update_task", {"project_id": -1}) == set()
+    assert await write_scope.tool_targets(1, "create_task", {"project_id": 0}) == set()
+    assert await write_scope.tool_targets(1, "create_note", {"project_id": 12}) == {12}
+
+
+async def test_reads_and_unknown_tools_land_nowhere():
+    assert await write_scope.tool_targets(1, "get_task", {"task_id": 5}) == set()
+    assert await write_scope.tool_targets(1, "no_such_tool", {"project_id": 5}) == set()
+
+
+def _patched(landing: set[int], readable: bool = True):
+    return (
+        patch.object(write_scope, "tool_targets", AsyncMock(return_value=landing)),
+        patch("grimoire.services.access.can_read_project", AsyncMock(return_value=readable)),
+        patch.object(write_scope, "_titles", AsyncMock(return_value={2: "Home", 9: "Elsewhere"})),
+    )
+
+
+async def test_a_write_in_the_sessions_own_project_goes_ahead():
+    a, b, c = _patched({2})
+    with a, b, c:
+        out = await check_act(1, 2, tool="update_task", args={"task_id": 5})
+    assert out == {"allowed": True, "targets": [], "issue": False, "reason": ""}
+
+
+async def test_a_write_in_another_project_asks_and_says_how_to_file_an_issue_instead():
+    a, b, c = _patched({9})
+    with a, b, c:
+        out = await check_act(1, 2, tool="update_task", args={"task_id": 5})
+    assert out["allowed"] is False
+    assert out["targets"] == [{"id": 9, "title": "Elsewhere"}]
+    assert "Home (project 2)" in out["reason"]
+    assert "Elsewhere (project 9)" in out["reason"]
+    assert 'create_task(project_id=9, kind="issue")' in out["reason"]
+
+
+async def test_filing_an_issue_in_another_project_goes_ahead():
+    a, b, c = _patched({9})
+    with a, b, c:
+        out = await check_act(1, 2, tool="create_task", args={"project_id": 9, "kind": "issue"})
+    assert out["allowed"] is True and out["issue"] is True
+
+
+async def test_a_session_with_no_project_is_held_to_nothing():
+    a, b, c = _patched({9})
+    with a, b, c:
+        out = await check_act(1, 0, tool="update_task", args={"task_id": 5})
+    assert out["allowed"] is True
+
+
+async def test_a_project_the_caller_cannot_read_is_not_named():
+    a, b, c = _patched({9}, readable=False)
+    with a, b, c:
+        out = await check_act(1, 2, tool="update_task", args={"task_id": 5})
+    assert out == {"allowed": True, "targets": [], "issue": False, "reason": ""}
+
+
+async def test_a_directory_in_another_project_asks_in_the_words_of_the_act():
+    with patch("grimoire.services.access.can_read_project", AsyncMock(return_value=True)), \
+         patch.object(write_scope, "_titles", AsyncMock(return_value={2: "Home", 9: "Elsewhere"})):
+        out = await check_act(1, 2, target_project=9, act="git push in")
+        same = await check_act(1, 2, target_project=2, act="git push in")
+    assert out["allowed"] is False
+    assert "git push in Elsewhere (project 9)" in out["reason"]
+    assert same["allowed"] is True
+
+
+async def test_the_route_reads_the_session_the_tool_and_its_input():
+    from grimoire.app import create_app
+    from grimoire.routes import plugin
+
+    app = create_app()
+    answer = {"allowed": False, "targets": [{"id": 9, "title": "x"}], "issue": False, "reason": "r"}
+    check = AsyncMock(return_value=dict(answer))
+    url = ("/api/plugin/write-scope?project_id=2"
+           "&tool=mcp__plugin_grimoire_grimoire__update_task"
+           "&input=%7B%22task_id%22%3A5%7D&act=x")
+    async with app.test_request_context(url):
+        g.user = SimpleNamespace(id=1)
+        with patch.object(plugin.write_scope_svc, "check_act", check), \
+             patch.object(plugin, "can_read_project", AsyncMock(return_value=True)):
+            resp = await plugin.write_scope.__wrapped__()
+    body = await resp.get_json()
+    assert body["allowed"] is False and body["session_project"] == 2
+    _, kwargs = check.call_args
+    assert check.call_args.args == (1, 2)
+    assert kwargs["tool"] == "update_task" and kwargs["args"] == {"task_id": 5}
+
+
+async def test_the_route_resolves_a_target_repo_to_its_project():
+    from grimoire.app import create_app
+    from grimoire.routes import plugin
+
+    app = create_app()
+    check = AsyncMock(return_value={"allowed": True, "targets": [], "issue": False, "reason": ""})
+    url = "/api/plugin/write-scope?project_id=2&target_repo=https://forge.example/o/r.git&act=git%20push%20in"
+    async with app.test_request_context(url):
+        g.user = SimpleNamespace(id=1)
+        with patch.object(plugin.write_scope_svc, "check_act", check), \
+             patch.object(plugin, "can_read_project", AsyncMock(return_value=True)), \
+             patch.object(plugin.repo_bindings_svc, "resolve_project", AsyncMock(return_value=9)):
+            await plugin.write_scope.__wrapped__()
+    assert check.call_args.kwargs["target_project"] == 9
+    assert check.call_args.kwargs["act"] == "git push in"
+
+
+async def test_the_route_refuses_input_that_is_not_an_object():
+    from grimoire.app import create_app
+    from grimoire.routes import plugin
+
+    app = create_app()
+    async with app.test_request_context("/api/plugin/write-scope?tool=update_task&input=%5B1%5D"):
+        g.user = SimpleNamespace(id=1)
+        with patch.object(plugin, "can_read_project", AsyncMock(return_value=True)):
+            resp, status = await plugin.write_scope.__wrapped__()
+    assert status == 400