feat(canvas): agent scope + permission gating and honest attribution (#1800 slice 3)

tests/projects/test_canvas_integration.py

@@ -30,6 +30,17 @@ async def app_env(tmp_path):
     await snap._ensure_subscribed(p["id"])
 
     app = FastAPI()
+
+    # The gated canvas routes authorize via request.state.user_id / is_admin
+    # (set by the auth middleware in production). This bare test app has no
+    # middleware, so inject a fixed admin session to mirror the unchanged
+    # session-owner behavior the routes rely on.
+    @app.middleware("http")
+    async def _inject_session(request, call_next):
+        request.state.user_id = "admin"
+        request.state.is_admin = True
+        return await call_next(request)
+
     app.state.project_store = ps
     app.state.project_canvas_store = cs
     app.state.canvas_snapshotter = snap

tests/projects/test_canvas_mcp_tools.py

@@ -28,15 +28,25 @@ async def env(tmp_path):
     ctx = ct.CanvasToolContext(
         project_store=ps, canvas_store=cs, snapshotter=snap, data_root=data_root,
     )
-    yield p, ctx, ps
+
+    async def _grant_edit():
+        await ps._db.execute(
+            "UPDATE project_members SET can_edit_canvas = 1 "
+            "WHERE project_id = ? AND member_id = ?",
+            (p["id"], "agent-1"),
+        )
+        await ps._db.commit()
+
+    yield p, ctx, ps, _grant_edit
     await snap.stop()
     await cs.close()
     await ps.close()
 
 
 @pytest.mark.asyncio
 async def test_canvas_add_note_creates_element(env):
-    p, ctx, _ = env
+    p, ctx, _ps, grant = env
+    await grant()
     res = await ct.canvas_add_note(
         ctx, project_id=p["id"], agent_id="agent-1",
         text="agent-said-hello", x=10, y=20,
@@ -48,7 +58,8 @@ async def test_canvas_add_note_creates_element(env):
 
 @pytest.mark.asyncio
 async def test_canvas_add_text_creates_element(env):
-    p, ctx, _ = env
+    p, ctx, _ps, grant = env
+    await grant()
     res = await ct.canvas_add_text(
         ctx, project_id=p["id"], agent_id="agent-1",
         text="an idea", x=5, y=5, font_size=20,
@@ -62,7 +73,8 @@ async def test_canvas_add_text_creates_element(env):
 
 @pytest.mark.asyncio
 async def test_canvas_add_mermaid_and_flowchart_carry_source(env):
-    p, ctx, _ = env
+    p, ctx, _ps, grant = env
+    await grant()
     m = await ct.canvas_add_mermaid(
         ctx, project_id=p["id"], agent_id="agent-1",
         source="graph TD; A-->B", x=0, y=0,
@@ -79,7 +91,8 @@ async def test_canvas_add_mermaid_and_flowchart_carry_source(env):
 
 @pytest.mark.asyncio
 async def test_canvas_add_mindmap_edge_links_endpoints(env):
-    p, ctx, _ = env
+    p, ctx, _ps, grant = env
+    await grant()
     a = await ct.canvas_add_text(
         ctx, project_id=p["id"], agent_id="agent-1", text="a", x=0, y=0,
     )
@@ -100,7 +113,8 @@ async def test_canvas_add_mindmap_edge_links_endpoints(env):
 
 @pytest.mark.asyncio
 async def test_canvas_add_mindmap_edge_rejects_unknown_endpoint(env):
-    p, ctx, _ = env
+    p, ctx, _ps, grant = env
+    await grant()
     a = await ct.canvas_add_text(
         ctx, project_id=p["id"], agent_id="agent-1", text="a", x=0, y=0,
     )
@@ -112,32 +126,71 @@ async def test_canvas_add_mindmap_edge_rejects_unknown_endpoint(env):
 
 
 @pytest.mark.asyncio
-async def test_canvas_update_denied_without_permission(env):
-    p, ctx, _ = env
-    note = await ct.canvas_add_note(
+async def test_canvas_add_denied_without_permission(env):
+    p, ctx, _ps, _grant = env
+    res = await ct.canvas_add_note(
         ctx, project_id=p["id"], agent_id="agent-1",
         text="x", x=0, y=0,
     )
+    assert res["error"] == "permission_denied"
+
+
+@pytest.mark.asyncio
+async def test_canvas_list_denied_without_read(env):
+    p, ctx, _ps, grant = env
+    # can_edit_canvas only: the read checkbox stays off, so the in-process
+    # read path must be blocked even though the agent can write.
+    await grant()
+    res = await ct.canvas_list_elements(
+        ctx, project_id=p["id"], agent_id="agent-1"
+    )
+    assert res["error"] == "permission_denied"
+
+
+@pytest.mark.asyncio
+async def test_canvas_list_allowed_with_read(env):
+    p, ctx, ps, grant = env
+    await grant()
+    await ps._db.execute(
+        "UPDATE project_members SET can_read_canvas = 1 "
+        "WHERE project_id = ? AND member_id = ?",
+        (p["id"], "agent-1"),
+    )
+    await ps._db.commit()
+    el = await ctx.canvas_store.add_element(
+        project_id=p["id"], author_kind="user", author_id="u",
+        element={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1, "payload": {"text": "x"}},
+    )
+    res = await ct.canvas_list_elements(
+        ctx, project_id=p["id"], agent_id="agent-1"
+    )
+    assert "elements" in res
+    assert [e["id"] for e in res["elements"]] == [el["id"]]
+
+
+@pytest.mark.asyncio
+async def test_canvas_update_denied_without_permission(env):
+    p, ctx, _ps, _grant = env
+    # An agent without the edit flag cannot mutate an existing element.
+    el = await ctx.canvas_store.add_element(
+        project_id=p["id"], author_kind="user", author_id="u",
+        element={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1, "payload": {"text": "x"}},
+    )
     res = await ct.canvas_update_element(
         ctx, project_id=p["id"], agent_id="agent-1",
-        element_id=note["element"]["id"], patch={"x": 99},
+        element_id=el["id"], patch={"x": 99},
     )
     assert res["error"] == "permission_denied"
 
 
 @pytest.mark.asyncio
 async def test_canvas_update_succeeds_with_permission(env):
-    p, ctx, ps = env
+    p, ctx, ps, grant = env
+    await grant()
     note = await ct.canvas_add_note(
         ctx, project_id=p["id"], agent_id="agent-1",
         text="x", x=0, y=0,
     )
-    await ps._db.execute(
-        "UPDATE project_members SET can_edit_canvas = 1 "
-        "WHERE project_id = ? AND member_id = ?",
-        (p["id"], "agent-1"),
-    )
-    await ps._db.commit()
     res = await ct.canvas_update_element(
         ctx, project_id=p["id"], agent_id="agent-1",
         element_id=note["element"]["id"], patch={"x": 99},
@@ -147,7 +200,8 @@ async def test_canvas_update_succeeds_with_permission(env):
 
 @pytest.mark.asyncio
 async def test_canvas_get_snapshot_png_writes_file(env):
-    p, ctx, _ = env
+    p, ctx, _ps, grant = env
+    await grant()
     await ct.canvas_add_note(
         ctx, project_id=p["id"], agent_id="agent-1", text="x", x=0, y=0,
     )

tests/projects/test_canvas_store.py

@@ -73,10 +73,17 @@ async def test_user_can_add_ideas_board_kind(store, kind):
 
 @pytest.mark.parametrize("kind", ["text", "mermaid", "flowchart", "mindmap_edge"])
 @pytest.mark.asyncio
-async def test_agent_can_emit_ideas_board_kind(store, kind):
-    # Agents may paint the ideas-board kinds via the canvas tools (#68).
-    e = await store.add_element(
-        project_id="p",
+async def test_agent_can_emit_ideas_board_kind(store_with_member, kind):
+    # Agents may paint the ideas-board kinds via the canvas tools (#68). The
+    # store enforces can_edit_canvas, so the member needs the edit flag set.
+    cs, _ = store_with_member
+    await cs._db.execute(
+        "UPDATE project_members SET can_edit_canvas = 1 "
+        "WHERE project_id = 'p1' AND member_id = 'agent-1'"
+    )
+    await cs._db.commit()
+    e = await cs.add_element(
+        project_id="p1",
         element={"kind": kind, "x": 0, "y": 0, "w": 1, "h": 1, "payload": {}},
         author_kind="agent", author_id="agent-1",
     )

tests/projects/test_routes_canvas.py

@@ -31,6 +31,17 @@ async def client(tmp_path):
     await snap.start()
 
     app = FastAPI()
+
+    # The gated canvas routes authorize via request.state.user_id / is_admin
+    # (set by the auth middleware in production). This bare test app has no
+    # middleware, so inject a fixed admin session for every request to mirror
+    # the unchanged session-owner behavior the routes rely on.
+    @app.middleware("http")
+    async def _inject_session(request, call_next):
+        request.state.user_id = "admin"
+        request.state.is_admin = True
+        return await call_next(request)
+
     app.state.project_store = ps
     app.state.project_canvas_store = cs
     app.state.canvas_snapshotter = snap

tests/test_auth_middleware.py

@@ -289,8 +289,10 @@ def test_snapshot_tldr_allowed(self):
     def test_stream_allowed(self):
         assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/stream") is True
 
-    def test_update_element_patch_not_allowed(self):
-        assert _is_agent_canvas_path("PATCH", "/api/projects/proj-1/canvas/elements/el-1") is False
+    def test_update_element_patch_allowed(self):
+        # canvas_write-bound agents may PATCH an element (create + update +
+        # delete all live under canvas_write), so the route is on the allowlist.
+        assert _is_agent_canvas_path("PATCH", "/api/projects/proj-1/canvas/elements/el-1") is True
 
     def test_permissions_patch_not_allowed(self):
         assert _is_agent_canvas_path("PATCH", "/api/projects/proj-1/canvas/permissions/agent-1") is False