diff --git a/desktop/src/lib/projects.test.ts b/desktop/src/lib/projects.test.ts
index 19de43981..ba2f7e846 100644
--- a/desktop/src/lib/projects.test.ts
+++ b/desktop/src/lib/projects.test.ts
@@ -269,27 +269,27 @@ describe("projectsApi.members.remove", () => {
});
});
-describe("projectsApi.members.setLead", () => {
+describe("projectsApi.setLead (D7, exclusive project pointer)", () => {
it("returns result on 200", async () => {
- mockFetch({ ok: true, is_lead: true });
- const result = await projectsApi.members.setLead("p-1", "m-1", true);
- expect(result.is_lead).toBe(true);
+ mockFetch({ ok: true, lead_member_id: "m-1" });
+ const result = await projectsApi.setLead("p-1", "m-1");
+ expect(result.lead_member_id).toBe("m-1");
});
- it("patches with is_lead body", async () => {
- const fetchMock = vi.fn().mockResolvedValue({ ok: true, json: async () => ({ ok: true, is_lead: true }) });
+ it("patches the project lead endpoint with a member_id body", async () => {
+ const fetchMock = vi.fn().mockResolvedValue({ ok: true, json: async () => ({ ok: true, lead_member_id: null }) });
global.fetch = fetchMock;
- await projectsApi.members.setLead("p-1", "m-1", false);
+ await projectsApi.setLead("p-1", null);
const [url, opts] = fetchMock.mock.calls[0];
- expect(url).toBe("/api/projects/p-1/members/m-1/lead");
+ expect(url).toBe("/api/projects/p-1/lead");
expect(opts.method).toBe("PATCH");
const body = JSON.parse(opts.body);
- expect(body.is_lead).toBe(false);
+ expect(body.member_id).toBe(null);
});
it("throws on non-ok response", async () => {
mockFetchError(403, "forbidden");
- await expect(projectsApi.members.setLead("p-1", "m-1", true)).rejects.toThrow("403");
+ await expect(projectsApi.setLead("p-1", "m-1")).rejects.toThrow("403");
});
});
diff --git a/desktop/src/lib/projects.ts b/desktop/src/lib/projects.ts
index 6ff385133..51578dfd0 100644
--- a/desktop/src/lib/projects.ts
+++ b/desktop/src/lib/projects.ts
@@ -7,6 +7,7 @@ export type Project = {
created_by: string;
created_at: number;
updated_at: number;
+ lead_member_id?: string | null;
};
export type ProjectMember = {
@@ -18,6 +19,7 @@ export type ProjectMember = {
memory_seed: "none" | "snapshot" | "empty";
added_at: number;
can_edit_canvas?: boolean;
+ can_read_canvas?: boolean;
is_lead?: number;
};
@@ -155,6 +157,13 @@ export const projectsApi = {
http
(`/api/projects/${id}/archive`, { method: "POST" }),
remove: (id: string) =>
http(`/api/projects/${id}`, { method: "DELETE" }),
+ // D7: the Lead is an exclusive, project-level designation. Pass a member id
+ // to promote that member, or null to clear the lead.
+ setLead: (id: string, member_id: string | null) =>
+ http<{ ok: boolean; lead_member_id: string | null }>(
+ `/api/projects/${id}/lead`,
+ { method: "PATCH", body: JSON.stringify({ member_id }) },
+ ),
members: {
list: (pid: string) =>
@@ -171,11 +180,6 @@ export const projectsApi = {
}),
remove: (pid: string, member_id: string) =>
http<{ ok: boolean }>(`/api/projects/${pid}/members/${member_id}`, { method: "DELETE" }),
- setLead: (pid: string, member_id: string, is_lead: boolean) =>
- http<{ ok: boolean; is_lead: boolean }>(
- `/api/projects/${pid}/members/${member_id}/lead`,
- { method: "PATCH", body: JSON.stringify({ is_lead }) },
- ),
},
tasks: {
diff --git a/tests/projects/test_a2a.py b/tests/projects/test_a2a.py
index d6b6290c4..5d0235e07 100644
--- a/tests/projects/test_a2a.py
+++ b/tests/projects/test_a2a.py
@@ -370,13 +370,13 @@ async def enqueue_user_message(self, slug: str, msg: dict) -> None:
# ---------------------------------------------------------------------------
-# Lead agent — ensure_a2a_channel syncs settings.leads
+# Lead agent — ensure_a2a_channel syncs settings.leads from lead_member_id (D7)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
-async def test_ensure_populates_leads_from_is_lead_flag(stores):
- """When a member has is_lead=1, their name appears in settings.leads."""
+async def test_ensure_populates_leads_from_lead_member_id(stores):
+ """When a member is the project's lead_member_id, their name appears in settings.leads."""
project_store, channel_store = stores
p = await project_store.create_project(name="P", slug="lead-basic", created_by="u1")
@@ -386,7 +386,7 @@ async def test_ensure_populates_leads_from_is_lead_flag(stores):
await project_store.add_member(p["id"], coord_id, member_kind="native")
await project_store.add_member(p["id"], worker_id, member_kind="native")
- await project_store.set_member_lead(p["id"], coord_id, True)
+ await project_store.set_lead(p["id"], coord_id)
ch = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
@@ -396,7 +396,7 @@ async def test_ensure_populates_leads_from_is_lead_flag(stores):
@pytest.mark.asyncio
async def test_ensure_leads_empty_when_no_leads(stores):
- """settings.leads is an empty list when no member is marked as lead."""
+ """settings.leads is an empty list when no member is the lead."""
project_store, channel_store = stores
p = await project_store.create_project(name="P", slug="no-leads", created_by="u1")
@@ -411,7 +411,7 @@ async def test_ensure_leads_empty_when_no_leads(stores):
@pytest.mark.asyncio
async def test_ensure_updates_leads_on_subsequent_call(stores):
- """Toggling is_lead and calling ensure again updates settings.leads."""
+ """Changing lead_member_id and calling ensure again updates settings.leads."""
project_store, channel_store = stores
p = await project_store.create_project(name="P", slug="lead-update", created_by="u1")
@@ -422,18 +422,18 @@ async def test_ensure_updates_leads_on_subsequent_call(stores):
ch1 = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
assert ch1["settings"]["leads"] == []
- await project_store.set_member_lead(p["id"], coord_id, True)
+ await project_store.set_lead(p["id"], coord_id)
ch2 = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
assert ch2["settings"]["leads"] == ["coord"]
- await project_store.set_member_lead(p["id"], coord_id, False)
+ await project_store.set_lead(p["id"], None)
ch3 = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
assert ch3["settings"]["leads"] == []
@pytest.mark.asyncio
-async def test_ensure_multiple_leads(stores):
- """Multiple lead members all appear in settings.leads, sorted."""
+async def test_ensure_lead_is_exclusive(stores):
+ """Setting a second lead replaces the first; only one name is ever in settings.leads."""
project_store, channel_store = stores
p = await project_store.create_project(name="P", slug="multi-lead", created_by="u1")
@@ -443,17 +443,20 @@ async def test_ensure_multiple_leads(stores):
await project_store.add_member(p["id"], alpha_id, member_kind="native")
await project_store.add_member(p["id"], beta_id, member_kind="native")
- await project_store.set_member_lead(p["id"], alpha_id, True)
- await project_store.set_member_lead(p["id"], beta_id, True)
+ await project_store.set_lead(p["id"], alpha_id)
- ch = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
+ ch1 = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
+ assert ch1["settings"]["leads"] == ["alpha"]
- assert ch["settings"]["leads"] == ["alpha", "beta"]
+ # Promoting beta must atomically unset alpha.
+ await project_store.set_lead(p["id"], beta_id)
+ ch2 = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
+ assert ch2["settings"]["leads"] == ["beta"]
@pytest.mark.asyncio
async def test_ensure_lead_removed_drops_from_leads(stores):
- """Removing a lead member from the project naturally drops them from settings.leads."""
+ """Removing the lead member from the project naturally drops them from settings.leads."""
project_store, channel_store = stores
p = await project_store.create_project(name="P", slug="lead-drop", created_by="u1")
@@ -463,7 +466,7 @@ async def test_ensure_lead_removed_drops_from_leads(stores):
await project_store.add_member(p["id"], coord_id, member_kind="native")
await project_store.add_member(p["id"], worker_id, member_kind="native")
- await project_store.set_member_lead(p["id"], coord_id, True)
+ await project_store.set_lead(p["id"], coord_id)
ch1 = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
assert ch1["settings"]["leads"] == ["coord"]
@@ -473,3 +476,28 @@ async def test_ensure_lead_removed_drops_from_leads(stores):
ch2 = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
assert ch2["settings"]["leads"] == []
assert ch2["members"] == ["worker"]
+
+
+@pytest.mark.asyncio
+async def test_ensure_lead_follows_backfilled_column(stores):
+ """A legacy is_lead=1 flag is backfilled into lead_member_id on init and synced."""
+ project_store, channel_store = stores
+ p = await project_store.create_project(name="P", slug="lead-backfill", created_by="u1")
+
+ coord_id = "iiiijjjjjjjj"
+ config = _config(_agent("coord", coord_id))
+ await project_store.add_member(p["id"], coord_id, member_kind="native")
+
+ # Simulate pre-D7 data: a per-member is_lead flag, no project pointer yet.
+ await project_store._db.execute(
+ "UPDATE project_members SET is_lead = 1 WHERE project_id = ? AND member_id = ?",
+ (p["id"], coord_id),
+ )
+ await project_store._db.commit()
+ await project_store._backfill_lead_member_id()
+
+ refreshed = await project_store.get_project(p["id"])
+ assert refreshed["lead_member_id"] == coord_id
+
+ ch = await ensure_a2a_channel(channel_store, project_store, p["id"], config=config)
+ assert ch["settings"]["leads"] == ["coord"]
diff --git a/tests/projects/test_canvas_integration.py b/tests/projects/test_canvas_integration.py
index c7930bfb2..39b732ce7 100644
--- a/tests/projects/test_canvas_integration.py
+++ b/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
diff --git a/tests/projects/test_canvas_mcp_tools.py b/tests/projects/test_canvas_mcp_tools.py
index 4bc191c2d..d37b4bcc9 100644
--- a/tests/projects/test_canvas_mcp_tools.py
+++ b/tests/projects/test_canvas_mcp_tools.py
@@ -28,7 +28,16 @@ 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()
@@ -36,7 +45,8 @@ async def env(tmp_path):
@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,
)
diff --git a/tests/projects/test_canvas_store.py b/tests/projects/test_canvas_store.py
index 464be1a3f..bd772e946 100644
--- a/tests/projects/test_canvas_store.py
+++ b/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",
)
diff --git a/tests/projects/test_routes_a2a.py b/tests/projects/test_routes_a2a.py
index 1618080f3..2ce1d3c7b 100644
--- a/tests/projects/test_routes_a2a.py
+++ b/tests/projects/test_routes_a2a.py
@@ -110,13 +110,13 @@ async def test_project_delete_archives_a2a_channel(client):
# ---------------------------------------------------------------------------
-# Lead endpoint — PATCH /api/projects/{pid}/members/{mid}/lead
+# Lead endpoint — PATCH /api/projects/{pid}/lead (D7, exclusive designee)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_set_lead_updates_channel_settings(client):
- """PATCHing the lead flag resynchronises settings.leads in the A2A channel."""
+ """PATCHing the lead pointer resynchronises settings.leads in the A2A channel."""
agent_id, agent_name = await _test_agent_id(client)
pid = (await client.post("/api/projects", json={"name": "P", "slug": "ra2a-lead1"})).json()["id"]
await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": agent_id})
@@ -128,11 +128,11 @@ async def test_set_lead_updates_channel_settings(client):
# Promote to lead
res = await client.patch(
- f"/api/projects/{pid}/members/{agent_id}/lead",
- json={"is_lead": True},
+ f"/api/projects/{pid}/lead",
+ json={"member_id": agent_id},
)
assert res.status_code == 200
- assert res.json()["is_lead"] is True
+ assert res.json()["lead_member_id"] == agent_id
a2a_after = _a2a(await _list_channels(client, pid))
assert a2a_after is not None
@@ -140,29 +140,55 @@ async def test_set_lead_updates_channel_settings(client):
@pytest.mark.asyncio
-async def test_set_lead_false_removes_from_leads(client):
- """Unsetting is_lead removes the agent name from settings.leads."""
+async def test_set_lead_null_clears_from_leads(client):
+ """Clearing the lead pointer (member_id: null) removes the agent from settings.leads."""
agent_id, agent_name = await _test_agent_id(client)
pid = (await client.post("/api/projects", json={"name": "P", "slug": "ra2a-lead2"})).json()["id"]
await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": agent_id})
- await client.patch(f"/api/projects/{pid}/members/{agent_id}/lead", json={"is_lead": True})
+ await client.patch(f"/api/projects/{pid}/lead", json={"member_id": agent_id})
res = await client.patch(
- f"/api/projects/{pid}/members/{agent_id}/lead",
- json={"is_lead": False},
+ f"/api/projects/{pid}/lead",
+ json={"member_id": None},
)
assert res.status_code == 200
+ assert res.json()["lead_member_id"] is None
a2a = _a2a(await _list_channels(client, pid))
assert a2a is not None
assert agent_name not in (a2a["settings"].get("leads") or [])
+@pytest.mark.asyncio
+async def test_set_lead_exclusive_replaces_previous(client):
+ """Setting a second lead unsets the first; only one lead ever remains."""
+ a1, n1 = await _test_agent_id(client)
+ # A second distinct agent: mint via registry-independent member add is not
+ # possible (members must reference config agents), so drive exclusivity
+ # through two calls against the same test agent path by reusing the route
+ # with the same member after clearing — verify the pointer is a single cell.
+ pid = (await client.post("/api/projects", json={"name": "P", "slug": "ra2a-lead-excl"})).json()["id"]
+ await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": a1})
+
+ first = await client.patch(f"/api/projects/{pid}/lead", json={"member_id": a1})
+ assert first.status_code == 200
+ assert first.json()["lead_member_id"] == a1
+
+ # Clear then set again: still exactly one lead (the pointer cannot hold two).
+ cleared = await client.patch(f"/api/projects/{pid}/lead", json={"member_id": None})
+ assert cleared.json()["lead_member_id"] is None
+ second = await client.patch(f"/api/projects/{pid}/lead", json={"member_id": a1})
+ assert second.json()["lead_member_id"] == a1
+
+ project = (await client.get(f"/api/projects/{pid}")).json()
+ assert project["lead_member_id"] == a1
+
+
@pytest.mark.asyncio
async def test_set_lead_nonexistent_member_returns_404(client):
pid = (await client.post("/api/projects", json={"name": "P", "slug": "ra2a-lead3"})).json()["id"]
res = await client.patch(
- f"/api/projects/{pid}/members/nonexistent-id/lead",
- json={"is_lead": True},
+ f"/api/projects/{pid}/lead",
+ json={"member_id": "nonexistent-id"},
)
assert res.status_code == 404
diff --git a/tests/projects/test_routes_canvas.py b/tests/projects/test_routes_canvas.py
index eecf98a49..592b3fa05 100644
--- a/tests/projects/test_routes_canvas.py
+++ b/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
diff --git a/tests/test_auth_middleware.py b/tests/test_auth_middleware.py
index 3ff1bd078..798da6657 100644
--- a/tests/test_auth_middleware.py
+++ b/tests/test_auth_middleware.py
@@ -9,6 +9,7 @@
from tinyagentos.auth_middleware import (
AuthMiddleware,
+ _is_agent_canvas_path,
_is_exempt,
_is_loopback_client,
)
@@ -265,5 +266,127 @@ async def test_prepare_shutdown_denied_from_remote(self):
resp = await middleware.dispatch(req, call_next)
+ assert resp.status_code == 401
+ call_next.assert_not_awaited()
+
+
+class TestIsAgentCanvasPath:
+ def test_list_elements_get_allowed(self):
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/elements") is True
+
+ def test_create_element_post_allowed(self):
+ assert _is_agent_canvas_path("POST", "/api/projects/proj-1/canvas/elements") is True
+
+ def test_delete_element_allowed(self):
+ assert _is_agent_canvas_path("DELETE", "/api/projects/proj-1/canvas/elements/el-1") is True
+
+ def test_snapshot_png_allowed(self):
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshot.png") is True
+
+ def test_snapshot_tldr_allowed(self):
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshot.tldr") is True
+
+ def test_stream_allowed(self):
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/stream") is True
+
+ 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
+
+ def test_extra_path_segment_not_allowed(self):
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/elements/el-1/extra") is False
+
+ def test_wrong_method_not_allowed(self):
+ assert _is_agent_canvas_path("POST", "/api/projects/proj-1/canvas/elements/el-1") is False
+
+ def test_nested_element_path_not_allowed(self):
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/elements/x/y") is False
+
+ def test_single_element_get_not_allowed(self):
+ # There is no GET /elements/{id} route in the allowlist; only the
+ # collection GET and the DELETE of a single element are permitted.
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/elements/el-1") is False
+
+ def test_snapshot_dot_is_literal_not_wildcard(self):
+ # The dot in snapshot.png / snapshot.tldr must be a literal, not a regex
+ # wildcard: a near-miss like snapshotXpng must NOT slip through the
+ # agent-token allowlist onto a session-only surface.
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshotXpng") is False
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshot_png") is False
+ assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshotXtldr") is False
+
+
+class TestCanvasAgentTokenDispatch:
+ @pytest.mark.asyncio
+ async def test_canvas_list_elements_bearer_passes(self):
+ middleware = AuthMiddleware(app=MagicMock())
+ auth_mgr = _default_auth_mgr()
+ auth_mgr.validate_local_token.return_value = False
+ req = _request(
+ method="GET",
+ path="/api/projects/proj-1/canvas/elements",
+ headers={"authorization": "Bearer registry-jwt"},
+ auth_mgr=auth_mgr,
+ )
+ call_next = AsyncMock(return_value=JSONResponse({"elements": []}))
+
+ resp = await middleware.dispatch(req, call_next)
+
+ assert resp.status_code == 200
+ assert req.state.via == "registry_jwt_candidate"
+ call_next.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ async def test_canvas_delete_element_bearer_passes(self):
+ middleware = AuthMiddleware(app=MagicMock())
+ auth_mgr = _default_auth_mgr()
+ auth_mgr.validate_local_token.return_value = False
+ req = _request(
+ method="DELETE",
+ path="/api/projects/proj-1/canvas/elements/el-1",
+ headers={"authorization": "Bearer registry-jwt"},
+ auth_mgr=auth_mgr,
+ )
+ call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
+
+ resp = await middleware.dispatch(req, call_next)
+
+ assert resp.status_code == 200
+ assert req.state.via == "registry_jwt_candidate"
+ call_next.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ async def test_canvas_permissions_patch_requires_session(self):
+ middleware = AuthMiddleware(app=MagicMock())
+ req = _request(
+ method="PATCH",
+ path="/api/projects/proj-1/canvas/permissions/agent-1",
+ headers={"authorization": "Bearer registry-jwt"},
+ auth_mgr=_default_auth_mgr(),
+ )
+ call_next = AsyncMock()
+
+ resp = await middleware.dispatch(req, call_next)
+
+ assert resp.status_code == 401
+ call_next.assert_not_awaited()
+
+ @pytest.mark.asyncio
+ async def test_canvas_extra_segment_requires_session(self):
+ middleware = AuthMiddleware(app=MagicMock())
+ req = _request(
+ method="GET",
+ path="/api/projects/proj-1/canvas/elements/el-1/extra",
+ headers={"authorization": "Bearer registry-jwt"},
+ auth_mgr=_default_auth_mgr(),
+ )
+ call_next = AsyncMock()
+
+ resp = await middleware.dispatch(req, call_next)
+
assert resp.status_code == 401
call_next.assert_not_awaited()
\ No newline at end of file
diff --git a/tests/test_routes_agent_auth_requests.py b/tests/test_routes_agent_auth_requests.py
index 9ef0b6f65..797ae5a92 100644
--- a/tests/test_routes_agent_auth_requests.py
+++ b/tests/test_routes_agent_auth_requests.py
@@ -1,3 +1,4 @@
+import asyncio
import pytest
@@ -316,10 +317,617 @@ async def test_approve_bare_at_claim_falls_back_to_framework(
await grants.close()
-class TestAgentAuthRequestsGet:
+class TestCanvasScopeApproval:
+ """Canvas scopes require a project_id and follow the same narrow-not-widen
+ rules as project_tasks."""
+
+ @pytest.mark.asyncio
+ async def test_approve_canvas_scopes_with_project_succeeds(
+ self, client, monkeypatch, tmp_path
+ ):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-canvas.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-canvas.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-canvas.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-canvas")
+
+ record = await auth_store.create(
+ identity_claim="canvas-bot",
+ framework="canvas-cli",
+ requested_scopes=["canvas_read", "canvas_write"],
+ requested_skills=None,
+ reason="",
+ duration_secs=None,
+ project_id="prj-canvas-1",
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={
+ "granted_scopes": ["canvas_read", "canvas_write"],
+ "project_id": "prj-canvas-1",
+ },
+ )
+ assert resp.status_code == 200, resp.text
+ data = resp.json()
+ assert data["status"] == "accepted"
+ assert data["canonical_id"]
+
+ agents = await registry.list_all()
+ assert len(agents) == 1
+ agent_grants = await grants.list_grants(agents[0]["canonical_id"])
+ assert {g["scope"] for g in agent_grants} == {"canvas_read", "canvas_write"}
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
+
+ @pytest.mark.asyncio
+ async def test_approve_canvas_scopes_without_project_is_rejected(
+ self, client, monkeypatch, tmp_path
+ ):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-canvas-np.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-canvas-np.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-canvas-np.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-canvas-np")
+
+ record = await auth_store.create(
+ identity_claim="canvas-bot",
+ framework="canvas-cli",
+ requested_scopes=["canvas_read", "canvas_write"],
+ requested_skills=None,
+ reason="",
+ duration_secs=None,
+ project_id="prj-canvas-1",
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={"granted_scopes": ["canvas_read", "canvas_write"]},
+ )
+ assert resp.status_code == 400, resp.text
+ assert "project_id" in resp.text
+
+ # No agent should have been registered by the rejected approval.
+ assert await registry.list_all() == []
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
+
+ @pytest.mark.asyncio
+ async def test_approve_canvas_scopes_with_blank_project_is_rejected(
+ self, client, monkeypatch, tmp_path
+ ):
+ # A blank or whitespace project_id is not a real binding: it must fail
+ # closed exactly like a missing one, or a downstream truthy check would
+ # treat the token as unbound and allow cross-project canvas access.
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ for blank in ("", " "):
+ registry = AgentRegistryStore(tmp_path / f"reg-canvas-blank-{len(blank)}.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / f"auth-canvas-blank-{len(blank)}.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / f"grants-canvas-blank-{len(blank)}.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(
+ tmp_path / f"keys-canvas-blank-{len(blank)}"
+ )
+
+ record = await auth_store.create(
+ identity_claim="canvas-bot",
+ framework="canvas-cli",
+ requested_scopes=["canvas_write"],
+ requested_skills=None,
+ reason="",
+ duration_secs=None,
+ project_id="prj-canvas-1",
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={"granted_scopes": ["canvas_write"], "project_id": blank},
+ )
+ assert resp.status_code == 400, (blank, resp.text)
+ assert "project_id" in resp.text
+ # The blank-project approval must not register an agent.
+ assert await registry.list_all() == []
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
+
+ @pytest.mark.asyncio
+ async def test_approve_cannot_widen_canvas_scopes(
+ self, client, monkeypatch, tmp_path
+ ):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-canvas-widen.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-canvas-widen.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-canvas-widen.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-canvas-widen")
+
+ record = await auth_store.create(
+ identity_claim="canvas-bot",
+ framework="canvas-cli",
+ requested_scopes=["canvas_read"],
+ requested_skills=None,
+ reason="",
+ duration_secs=None,
+ project_id="prj-canvas-1",
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={
+ "granted_scopes": ["canvas_read", "canvas_write"],
+ "project_id": "prj-canvas-1",
+ },
+ )
+ assert resp.status_code == 400, resp.text
+ assert "subset of the requested scopes" in resp.text
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
+
+ @pytest.mark.asyncio
+ async def test_approve_canvas_scopes_cannot_widen_by_adding_project_tasks(
+ self, client, monkeypatch, tmp_path
+ ):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-canvas-mix.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-canvas-mix.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-canvas-mix.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-canvas-mix")
+
+ record = await auth_store.create(
+ identity_claim="canvas-bot",
+ framework="canvas-cli",
+ requested_scopes=["canvas_read"],
+ requested_skills=None,
+ reason="",
+ duration_secs=None,
+ project_id="prj-canvas-1",
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={
+ "granted_scopes": ["canvas_read", "project_tasks"],
+ "project_id": "prj-canvas-1",
+ },
+ )
+ assert resp.status_code == 400, resp.text
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
@pytest.mark.asyncio
async def test_get_unknown_id_returns_404(self, client, monkeypatch):
store = _FakeAuthRequestsStore()
monkeypatch.setattr(client._transport.app.state, "auth_requests", store)
resp = await client.get("/api/agents/auth-requests/nonexistent123")
assert resp.status_code == 404
+
+
+class TestHandleSetOnApprove:
+ """Slice 7: approve sets the registry handle from the sanitized identity
+ claim; the new agent then passes the a2a bus 'no handle' gate."""
+
+ @pytest.mark.asyncio
+ async def test_approve_sets_handle_on_registry(self, client, monkeypatch, tmp_path):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-handle.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-handle.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-handle.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-handle")
+
+ record = await auth_store.create(
+ identity_claim="@taOSmd-dev",
+ framework="openclaw",
+ requested_scopes=["memory_read"],
+ requested_skills=None,
+ reason="",
+ duration_secs=None,
+ project_id=None,
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={"granted_scopes": ["memory_read"]},
+ )
+ assert resp.status_code == 200, resp.text
+
+ agents = await registry.list_all()
+ assert len(agents) == 1
+ assert agents[0]["handle"] == "taosmd-dev"
+
+ handle = (agents[0].get("handle") or "").strip()
+ assert handle, "handle must be set so the a2a 'no handle' gate passes"
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
+
+
+class TestHandleCollisionActiveRejects:
+ """Slice 7: a handle collision with an ACTIVE identity returns 409 and
+ leaves the auth request PENDING so the approver can pick another variant."""
+
+ @pytest.mark.asyncio
+ async def test_409_when_handle_collides_with_active(
+ self, client, monkeypatch, tmp_path
+ ):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-collide.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-collide.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-collide.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-collide")
+
+ existing = await registry.register(
+ framework="openclaw",
+ display_name="existing-agent",
+ user_id="user-existing",
+ origin="taos-deployed",
+ handle="taosmd-dev",
+ )
+
+ record = await auth_store.create(
+ identity_claim="@taOSmd-dev",
+ framework="openclaw",
+ requested_scopes=["memory_read"],
+ requested_skills=None,
+ reason="",
+ duration_secs=None,
+ project_id=None,
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={"granted_scopes": ["memory_read"]},
+ )
+ assert resp.status_code == 409, resp.text
+
+ pending = await auth_store.get(record["id"])
+ assert pending is not None
+ assert pending["status"] == "pending"
+
+ active_agents = await registry.list_all(status="active")
+ assert len(active_agents) == 1
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
+
+
+class TestHandleCollisionSuspendedAllowsReuse:
+ """Slice 7: if the previous holder of the handle is SUSPENDED, the handle
+ may be reused and approval succeeds."""
+
+ @pytest.mark.asyncio
+ async def test_approve_succeeds_after_handle_holder_suspended(
+ self, client, monkeypatch, tmp_path
+ ):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-suspended.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-suspended.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-suspended.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-suspended")
+
+ old_agent = await registry.register(
+ framework="openclaw",
+ display_name="old-agent",
+ user_id="user-old",
+ origin="taos-deployed",
+ handle="taosmd-dev",
+ )
+ await registry.set_status(
+ old_agent["canonical_id"], "suspended", actor="user-old"
+ )
+
+ record = await auth_store.create(
+ identity_claim="@taOSmd-dev",
+ framework="openclaw",
+ requested_scopes=["memory_read"],
+ requested_skills=None,
+ reason="",
+ duration_secs=None,
+ project_id=None,
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={"granted_scopes": ["memory_read"]},
+ )
+ assert resp.status_code == 200, resp.text
+
+ new_agents = await registry.list_all()
+ new_active = [a for a in new_agents if a["status"] == "active"]
+ assert len(new_active) == 1
+ assert new_active[0]["handle"] == "taosmd-dev"
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
+
+
+class TestPartialUniqueHandleIndex:
+ """The DB (not just the application pre-check) must reject a duplicate
+ active handle, so two concurrent approvals of the same identity_claim
+ cannot both become active with identical handles (a2a 'from' spoofing)."""
+
+ @pytest.mark.asyncio
+ async def test_index_blocks_two_active_with_same_handle(self, tmp_path):
+ from tinyagentos.agent_registry_store import AgentRegistryStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-idx.db")
+ await registry.init()
+
+ # Both born pending with the same handle (allowed: index only covers
+ # active + non-empty handle).
+ a = await registry.register(
+ framework="openclaw", display_name="a", origin="external-selfjoin",
+ handle="dup-handle",
+ )
+ b = await registry.register(
+ framework="openclaw", display_name="b", origin="external-selfjoin",
+ handle="dup-handle",
+ )
+ await registry.set_status(a["canonical_id"], "active")
+
+ # The second activation must be rejected by the partial unique index.
+ with pytest.raises(Exception) as exc:
+ await registry.set_status(b["canonical_id"], "active")
+ assert "ux_agent_active_handle" in str(exc.value) or "UNIQUE" in str(exc.value)
+
+ # And the loser must never be left active with the handle.
+ b_row = await registry.get(b["canonical_id"])
+ assert b_row["status"] == "pending"
+ await registry.close()
+
+
+class TestConcurrentApproveSameIdentity:
+ """Two approvals racing on the SAME identity_claim: exactly one active
+ agent with the handle set, the other returns 409 and leaves no extra
+ active (or pending) agent behind."""
+
+ @pytest.mark.asyncio
+ async def test_concurrent_approve_one_active_other_409(
+ self, client, monkeypatch, tmp_path
+ ):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-race.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-race.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-race.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-race")
+
+ # Two independent auth requests for the SAME identity_claim. The
+ # per-request lock does NOT protect this case (different request ids),
+ # so this exercises the unique-index race directly.
+ r1 = await auth_store.create(
+ identity_claim="@taOSmd-dev", framework="openclaw",
+ requested_scopes=["memory_read"], requested_skills=None, reason="",
+ duration_secs=None, project_id=None,
+ )
+ r2 = await auth_store.create(
+ identity_claim="@taOSmd-dev", framework="openclaw",
+ requested_scopes=["memory_read"], requested_skills=None, reason="",
+ duration_secs=None, project_id=None,
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ async def _approve(rid):
+ return await client.post(
+ f"/api/agents/auth-requests/{rid}/approve",
+ json={"granted_scopes": ["memory_read"]},
+ )
+
+ # Drive both approvals concurrently so neither pre-check sees an active
+ # handle yet; the unique index decides the winner.
+ resps = await asyncio.gather(_approve(r1["id"]), _approve(r2["id"]))
+
+ statuses = sorted(r.status_code for r in resps)
+ assert statuses == [200, 409], [r.status_code for r in resps]
+
+ active = await registry.list_all(status="active")
+ assert len(active) == 1, active
+ # The winner's handle is set and non-empty.
+ assert active[0]["handle"] == "taosmd-dev"
+ assert active[0]["handle"]
+
+ # The loser is not left behind as an active or pending agent.
+ assert len(await registry.list_all()) == 1
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
+
+ @pytest.mark.asyncio
+ async def test_approved_active_handle_is_nonempty(
+ self, client, monkeypatch, tmp_path
+ ):
+ from tinyagentos.agent_registry_store import (
+ AgentRegistryStore,
+ load_or_create_signing_keypair,
+ )
+ from tinyagentos.auth_requests_store import AuthRequestsStore
+ from tinyagentos.agent_grants_store import AgentGrantsStore
+
+ registry = AgentRegistryStore(tmp_path / "reg-nonempty.db")
+ await registry.init()
+ auth_store = AuthRequestsStore(tmp_path / "auth-nonempty.db")
+ await auth_store.init()
+ grants = AgentGrantsStore(tmp_path / "grants-nonempty.db")
+ await grants.init()
+ priv, pub = load_or_create_signing_keypair(tmp_path / "keys-nonempty")
+
+ record = await auth_store.create(
+ identity_claim="grok-bot", framework="grok",
+ requested_scopes=["memory_read"], requested_skills=None, reason="",
+ duration_secs=None, project_id=None,
+ )
+
+ monkeypatch.setattr(client._transport.app.state, "agent_registry", registry)
+ monkeypatch.setattr(client._transport.app.state, "auth_requests", auth_store)
+ monkeypatch.setattr(client._transport.app.state, "agent_grants", grants)
+ monkeypatch.setattr(
+ client._transport.app.state, "agent_registry_keypair", (priv, pub)
+ )
+
+ resp = await client.post(
+ f"/api/agents/auth-requests/{record['id']}/approve",
+ json={"granted_scopes": ["memory_read"]},
+ )
+ assert resp.status_code == 200, resp.text
+
+ active = await registry.list_all(status="active")
+ assert len(active) == 1
+ # Never left active with an empty handle (the orphan-window bug).
+ assert active[0]["handle"]
+ assert active[0]["handle"] == "grok-bot"
+
+ await registry.close()
+ await auth_store.close()
+ await grants.close()
diff --git a/tests/test_routes_project_canvas.py b/tests/test_routes_project_canvas.py
index 220912054..80e16f006 100644
--- a/tests/test_routes_project_canvas.py
+++ b/tests/test_routes_project_canvas.py
@@ -3,27 +3,32 @@
from __future__ import annotations
import asyncio
-import json
-from pathlib import Path
+from types import SimpleNamespace
from unittest.mock import patch, AsyncMock
import pytest
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
+
+from tinyagentos.agent_registry_store import mint_registry_token
@pytest.fixture(autouse=True)
def _ensure_canvas_store(client, tmp_path_factory):
- """Initialize project_canvas_store if the lifespan didn't run (test client)."""
+ """Initialize project_canvas_store if the lifespan didn't run (test client).
+
+ Point the canvas store at the same projects.db the project_store uses, so
+ the per-project canvas permission live in the shared project_members table
+ (the store-level edit check reads from there, exactly as in production).
+ """
store = client._transport.app.state.project_canvas_store
if store._db is not None:
try:
asyncio.get_event_loop().run_until_complete(store.close())
except Exception:
pass
- # Use a fresh DB per test session via tmp_path. BaseStore reads self.db_path
- # (a Path) in init(); the previous override set a non-existent `_db_path`
- # string attr, so init() silently fell back to the production canvas DB.
- tmp_dir = tmp_path_factory.mktemp("canvas_test")
- store.db_path = tmp_dir / "test_projects.db"
+ ps = client._transport.app.state.project_store
+ store.db_path = ps.db_path
asyncio.get_event_loop().run_until_complete(store.init())
yield
try:
@@ -368,8 +373,11 @@ async def test_snapshot_tldr_requires_snapshotter(client):
@pytest.mark.asyncio
async def test_set_permission_member_not_found_returns_404(client):
+ resp = await client.post("/api/projects", json={"name": "perm", "slug": "perm"})
+ assert resp.status_code == 200, resp.text
+ pid = resp.json()["id"]
resp = await client.patch(
- "/api/projects/proj-1/canvas/permissions/agent-1",
+ f"/api/projects/{pid}/canvas/permissions/agent-1",
json={"can_edit_canvas": True},
)
assert resp.status_code == 404
@@ -389,3 +397,732 @@ async def test_canvas_stream_endpoint_exists(client):
app = client._transport.app
paths = [r.path for r in app.routes]
assert "/api/projects/{project_id}/canvas/stream" in paths
+
+
+# ---------------------------------------------------------------------------
+# Finding 2 (adversarial review): session users must be gated by project
+# visibility (owner/admin). A non-owner collapses into the same existence-hiding
+# 404 the agent path uses (D3 matrix); owner/admin stays allowed, and an
+# allowed session write is still attributed to the verified user.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+class TestSessionCanvasOwnerGating:
+ async def test_owner_session_write_allowed(self, ctx):
+ pid = await _new_project(ctx, "owner-gate-write")
+ resp = await ctx.client.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "owned"}},
+ )
+ assert resp.status_code == 201, resp.text
+
+ async def test_non_owner_session_list_is_404(self, ctx):
+ pid = await _new_project(ctx, "nonowner-gate")
+ async with _non_owner_client(ctx.app) as other:
+ resp = await other.get(f"/api/projects/{pid}/canvas/elements")
+ assert resp.status_code == 404
+ assert resp.status_code != 403
+ assert resp.status_code != 200
+
+ async def test_non_owner_session_write_is_404(self, ctx):
+ pid = await _new_project(ctx, "nonowner-gate-write")
+ async with _non_owner_client(ctx.app) as other:
+ resp = await other.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "x"}},
+ )
+ assert resp.status_code == 404
+ assert resp.status_code != 403
+
+
+# ---------------------------------------------------------------------------
+# Slice 3: agent scope + per-project canvas permission gating (D3 matrix)
+# plus honest attribution and the uniform actor stamp (D4).
+# ---------------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def ctx(client):
+ """Reuse the session-admin `client` app and init the agent registry +
+ grants stores so canvas agent tokens can be minted against real stores."""
+ app = client._transport.app
+ for attr in ("agent_registry", "agent_grants"):
+ store = getattr(app.state, attr)
+ if store._db is None:
+ await store.init()
+ # The test client bypasses the lifespan, so app.state.project_event_broker
+ # is unset; point it at the broker the canvas store already publishes to.
+ if getattr(app.state, "project_event_broker", None) is None:
+ app.state.project_event_broker = app.state.project_canvas_store._broker
+ uid = app.state.auth.find_user("admin")["id"]
+ yield SimpleNamespace(client=client, app=app, uid=uid)
+ for attr in ("agent_registry", "agent_grants"):
+ store = getattr(app.state, attr)
+ if store._db is not None:
+ await store.close()
+
+
+def _bare(app):
+ """Cookieless client so requests carry only the Bearer header."""
+ return AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
+
+
+def _hdr(token):
+ return {"Authorization": f"Bearer {token}"}
+
+
+async def _new_project(ctx, slug):
+ resp = await ctx.client.post("/api/projects", json={"name": slug, "slug": slug})
+ assert resp.status_code == 200, resp.text
+ return resp.json()["id"]
+
+
+def _non_owner_client(app):
+ """A non-owner, non-admin session client for a second human user.
+
+ The project under test is owned by the admin `client` fixture user, so a
+ session minted for this user is an "other" human per the D3 matrix and must
+ collapse into an existence-hiding 404 on the canvas.
+ """
+ auth = app.state.auth
+ code = auth.add_user_invite("other", "admin")
+ rec = auth.complete_invite(
+ "other", code, "Other User", "o@example.com", "password123"
+ )
+ token = auth.create_session(user_id=rec["id"], long_lived=True)
+ return AsyncClient(
+ transport=ASGITransport(app=app),
+ base_url="http://test",
+ cookies={"taos_session": token},
+ )
+
+
+async def _add_member(ctx, pid, member_id):
+ await ctx.app.state.project_store.add_member(pid, member_id, member_kind="native")
+
+
+async def _mint_agent(ctx, project_id, scopes):
+ registry = ctx.app.state.agent_registry
+ grants = ctx.app.state.agent_grants
+ priv, _pub = ctx.app.state.agent_registry_keypair
+ rec = await registry.register(
+ framework="grok",
+ display_name="Grok",
+ origin="external-selfjoin",
+ handle="@grok",
+ )
+ cid = rec["canonical_id"]
+ await registry.set_status(cid, "active")
+ for scope in scopes:
+ await grants.add_grant(cid, scope, project_id=project_id)
+ token = mint_registry_token(
+ cid, priv, user_id="u", framework="grok", project_id=project_id
+ )
+ return cid, token
+
+
+async def _grant_canvas(ctx, pid, agent_id, *, read=None, edit=None):
+ body = {}
+ if read is not None:
+ body["can_read_canvas"] = read
+ if edit is not None:
+ body["can_edit_canvas"] = edit
+ resp = await ctx.client.patch(
+ f"/api/projects/{pid}/canvas/permissions/{agent_id}", json=body
+ )
+ assert resp.status_code == 200, resp.text
+ return resp
+
+
+@pytest.mark.asyncio
+class TestAgentCanvasReadGating:
+ async def test_read_allowed_with_scope_and_flag(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/elements", headers=_hdr(token)
+ )
+ assert resp.status_code == 200
+
+ async def test_read_without_scope_is_403(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ # canvas_write only: read scope is missing.
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/elements", headers=_hdr(token)
+ )
+ assert resp.status_code == 403
+
+ async def test_read_without_flag_is_403(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ # No can_read_canvas grant.
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/elements", headers=_hdr(token)
+ )
+ assert resp.status_code == 403
+
+@pytest.mark.asyncio
+class TestAgentCanvasWriteGating:
+ async def test_write_allowed_with_scope_and_flag(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, edit=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "a"}},
+ headers=_hdr(token),
+ )
+ assert resp.status_code == 201, resp.text
+
+ async def test_write_without_scope_is_403(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ # canvas_read only: write scope is missing.
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, edit=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "a"}},
+ headers=_hdr(token),
+ )
+ assert resp.status_code == 403
+
+ async def test_write_without_flag_is_403(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ # No can_edit_canvas grant.
+ async with _bare(ctx.app) as bare:
+ resp = await bare.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "a"}},
+ headers=_hdr(token),
+ )
+ assert resp.status_code == 403
+
+ async def test_write_flag_does_not_grant_read(self, ctx):
+ """The edit flag (and canvas_write scope) must NOT satisfy a read."""
+ pid = await _new_project(ctx, "alpha")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, edit=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/elements", headers=_hdr(token)
+ )
+ assert resp.status_code == 403
+
+
+@pytest.mark.asyncio
+class TestAgentCanvasCrossProject:
+ async def test_different_project_is_404(self, ctx):
+ pid_a = await _new_project(ctx, "alpha")
+ pid_b = await _new_project(ctx, "bravo")
+ cid, token_a = await _mint_agent(ctx, pid_a, ("canvas_read",))
+ await _add_member(ctx, pid_a, cid)
+ await _grant_canvas(ctx, pid_a, cid, read=True)
+ async with _bare(ctx.app) as bare:
+ listing = await bare.get(
+ f"/api/projects/{pid_b}/canvas/elements", headers=_hdr(token_a)
+ )
+ assert listing.status_code == 404
+
+
+@pytest.mark.asyncio
+class TestCanvasAttribution:
+ async def test_agent_write_carry_agent_fields(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, edit=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "a"}},
+ headers=_hdr(token),
+ )
+ assert resp.status_code == 201, resp.text
+ el = resp.json()["element"]
+ assert el["author_kind"] == "agent"
+ assert el["author_id"] == cid
+
+ async def test_user_write_attributed_to_user_not_system(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ resp = await ctx.client.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "a"}},
+ )
+ assert resp.status_code == 201, resp.text
+ el = resp.json()["element"]
+ assert el["author_kind"] == "user"
+ assert el["author_id"] == ctx.uid
+ assert el["author_id"] != "system"
+
+
+@pytest.mark.asyncio
+class TestCanvasEventActor:
+ """D4: a uniform actor object {kind, id} is stamped on every canvas.* event."""
+
+ async def _capture(self, ctx, pid, coro):
+ broker = ctx.app.state.project_event_broker
+ queue = await broker.subscribe(pid)
+ # Drain any replayed events from earlier actions in this test so we
+ # observe only the event produced by the action under test.
+ while True:
+ try:
+ queue.get_nowait()
+ except asyncio.QueueEmpty:
+ break
+ await coro()
+ return await asyncio.wait_for(queue.get(), timeout=5.0)
+
+ async def test_actor_on_create_event(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+
+ async def action():
+ await ctx.client.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "a"}},
+ )
+
+ ev = await self._capture(ctx, pid, action)
+ assert ev.kind == "canvas.element_added"
+ assert ev.payload["actor"] == {"kind": "user", "id": ctx.uid}
+
+ async def test_actor_on_delete_event(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ created = await ctx.client.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "a"}},
+ )
+ eid = created.json()["element"]["id"]
+
+ async def action():
+ await ctx.client.delete(
+ f"/api/projects/{pid}/canvas/elements/{eid}"
+ )
+
+ ev = await self._capture(ctx, pid, action)
+ assert ev.kind == "canvas.element_deleted"
+ assert ev.payload["actor"] == {"kind": "user", "id": ctx.uid}
+
+ async def test_actor_on_permission_change_event(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ cid, _token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+
+ async def action():
+ await ctx.client.patch(
+ f"/api/projects/{pid}/canvas/permissions/{cid}",
+ json={"can_edit_canvas": True},
+ )
+
+ ev = await self._capture(ctx, pid, action)
+ assert ev.kind == "canvas.permission_changed"
+ assert ev.payload["actor"] == {"kind": "user", "id": ctx.uid}
+ assert ev.payload["can_edit_canvas"] is True
+
+
+@pytest.mark.asyncio
+class TestPermissionPatchExtendsFlags:
+ async def test_patch_can_set_read_flag_only(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ cid, _token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ resp = await ctx.client.patch(
+ f"/api/projects/{pid}/canvas/permissions/{cid}",
+ json={"can_read_canvas": True},
+ )
+ assert resp.status_code == 200, resp.text
+ assert resp.json()["can_read_canvas"] is True
+ assert resp.json()["can_edit_canvas"] is False
+ members = await ctx.app.state.project_store.list_members(pid)
+ me = next(m for m in members if m["member_id"] == cid)
+ assert me["can_read_canvas"] == 1
+ assert me["can_edit_canvas"] == 0
+
+ async def test_patch_can_set_edit_flag_only(self, ctx):
+ pid = await _new_project(ctx, "alpha")
+ cid, _token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ resp = await ctx.client.patch(
+ f"/api/projects/{pid}/canvas/permissions/{cid}",
+ json={"can_edit_canvas": True},
+ )
+ assert resp.status_code == 200, resp.text
+ assert resp.json()["can_edit_canvas"] is True
+ assert resp.json()["can_read_canvas"] is False
+
+ async def test_patch_rejects_agent_token(self, ctx):
+ """The permissions PATCH is owner/admin only; an agent token must not
+ reach it (the middleware leaves it off the agent allowlist -> 401)."""
+ pid = await _new_project(ctx, "alpha")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.patch(
+ f"/api/projects/{pid}/canvas/permissions/{cid}",
+ json={"can_edit_canvas": True},
+ headers=_hdr(token),
+ )
+ assert resp.status_code == 401
+
+
+# ---------------------------------------------------------------------------
+# Slice 4: the SSE stream and the snapshot endpoints are gated by canvas_read
+# for agent principals, and the keepalive tick re-checks the live read flag so
+# a revoked agent cannot keep a long-lived stream open (lead-agent-identity
+# design, Slice 4 + Edge cases).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+class TestCanvasStreamAgentGating:
+ async def _connect_status(self, ctx, pid, token):
+ """Open the SSE route directly so we never read the infinite body.
+
+ A denied agent either raises HTTPException (missing scope) or returns a
+ JSONResponse (flag off); an allowed agent returns a StreamingResponse.
+ """
+ from fastapi import HTTPException as _HTTPExc
+ from starlette.responses import StreamingResponse as _SR
+ from starlette.responses import JSONResponse as _JR
+ from tinyagentos.routes.project_canvas import canvas_stream
+
+ try:
+ resp = await canvas_stream(pid, _stream_req(ctx.app, token=token))
+ except _HTTPExc as exc:
+ return exc.status_code
+ if isinstance(resp, _SR):
+ return 200
+ if isinstance(resp, _JR):
+ return resp.status_code
+ return None
+
+ async def test_stream_connect_allowed_with_read_scope_and_flag(self, ctx):
+ pid = await _new_project(ctx, "stream-read")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ assert await self._connect_status(ctx, pid, token) == 200
+
+ async def test_stream_connect_without_scope_is_403(self, ctx):
+ pid = await _new_project(ctx, "stream-noscope")
+ # canvas_write only: the read scope is missing.
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ assert await self._connect_status(ctx, pid, token) == 403
+
+ async def test_stream_connect_without_flag_is_403(self, ctx):
+ pid = await _new_project(ctx, "stream-noflag")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ # can_read_canvas flag is off.
+ assert await self._connect_status(ctx, pid, token) == 403
+
+ async def test_stream_closes_after_read_flag_cleared(self, ctx):
+ """Slice 4 edge case: an agent principal's open stream must close once
+ its can_read_canvas flag is cleared, bounded by the keepalive interval."""
+ from tinyagentos.projects.events import ProjectEvent
+ from tinyagentos.routes.project_canvas import canvas_stream
+
+ pid = await _new_project(ctx, "stream-revoke")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ resp = await canvas_stream(pid, _stream_req(ctx.app, token=token))
+ gen = resp.body_iterator
+ broker = ctx.app.state.project_event_broker
+
+ # A published event lets the generator emit a data frame immediately,
+ # proving the agent principal's stream is live.
+ async def _emit(n):
+ await broker.publish(
+ pid, ProjectEvent(kind="canvas.element_added", payload={"n": n})
+ )
+
+ await _emit(1)
+ frame = await asyncio.wait_for(gen.__anext__(), timeout=5)
+ assert frame.startswith("data: ")
+ # Revoke read access while the stream is still open. The permissions
+ # PATCH publishes a canvas.permission_changed event that the live stream
+ # emits (proving liveness through events), then the next keepalive tick
+ # re-checks the flag and closes the stream.
+ await _grant_canvas(ctx, pid, cid, read=False)
+ closed = False
+ try:
+ while True:
+ await asyncio.wait_for(gen.__anext__(), timeout=12)
+ except StopAsyncIteration:
+ closed = True
+ assert closed
+
+
+@pytest.mark.asyncio
+class TestCanvasSnapshotAgentGating:
+ async def test_snapshot_png_allowed_with_read_scope_and_flag(self, ctx):
+ pid = await _new_project(ctx, "snap-read")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/snapshot.png", headers=_hdr(token)
+ )
+ assert resp.status_code == 200
+ assert resp.headers.get("content-type") == "image/png"
+
+ async def test_snapshot_png_write_only_is_403(self, ctx):
+ """Snapshots are READ scope only: a canvas_write-only token is denied."""
+ pid = await _new_project(ctx, "snap-wo")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/snapshot.png", headers=_hdr(token)
+ )
+ assert resp.status_code == 403
+
+ async def test_snapshot_png_without_flag_is_403(self, ctx):
+ pid = await _new_project(ctx, "snap-noflag")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/snapshot.png", headers=_hdr(token)
+ )
+ assert resp.status_code == 403
+
+ async def test_snapshot_tldr_write_only_is_403(self, ctx):
+ """Snapshots are READ scope only, even for the tldr endpoint."""
+ pid = await _new_project(ctx, "snap-tldr-wo")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/snapshot.tldr", headers=_hdr(token)
+ )
+ assert resp.status_code == 403
+
+ async def test_snapshot_tldr_allowed_with_read_scope_and_flag(self, ctx):
+ pid = await _new_project(ctx, "snap-tldr-read")
+ snap = ctx.app.state.canvas_snapshotter
+ if snap is None:
+ pytest.skip("canvas_snapshotter not available; needs container backend")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_read",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, read=True)
+ async with _bare(ctx.app) as bare:
+ resp = await bare.get(
+ f"/api/projects/{pid}/canvas/snapshot.tldr", headers=_hdr(token)
+ )
+ assert resp.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# Slice 4 regression: session owner/admin behavior on stream + snapshots is
+# unchanged by the agent gating.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+class TestCanvasStreamSnapshotSessionUnchanged:
+ async def test_owner_stream_connect_allowed(self, ctx):
+ from tinyagentos.routes.project_canvas import canvas_stream
+
+ pid = await _new_project(ctx, "owner-stream")
+ resp = await canvas_stream(
+ pid, _stream_req(ctx.app, user_id=ctx.uid, is_admin=True)
+ )
+ from starlette.responses import StreamingResponse as _SR
+
+ assert isinstance(resp, _SR)
+
+ async def test_owner_snapshot_png_allowed(self, ctx):
+ pid = await _new_project(ctx, "owner-snap")
+ resp = await ctx.client.get(
+ f"/api/projects/{pid}/canvas/snapshot.png"
+ )
+ assert resp.status_code == 200
+
+ async def test_session_gating_uses_project_visibility(self, ctx):
+ """A non-owner human session still collapses into 404 (D3 matrix),
+ so agent gating did not change session semantics."""
+ pid = await _new_project(ctx, "nonowner-snap")
+ async with _non_owner_client(ctx.app) as other:
+ resp = await other.get(
+ f"/api/projects/{pid}/canvas/snapshot.png"
+ )
+ assert resp.status_code == 404
+
+
+def _stream_req(app, *, token=None, user_id=None, is_admin=False):
+ """Build a minimal Starlette Request wired to the real app.state so the
+ canvas stream route can be invoked directly (no infinite-body client read)."""
+ from starlette.requests import Request as _StarletteRequest
+
+ headers = []
+ if token is not None:
+ headers.append((b"authorization", f"Bearer {token}".encode()))
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/x",
+ "query_string": b"",
+ "headers": headers,
+ "app": app,
+ }
+ req = _StarletteRequest(scope)
+ req.state.user_id = user_id
+ req.state.is_admin = is_admin
+
+ # is_disconnected() cancels its scope and awaits receive; an immediate,
+ # non-awaiting receive returns a normal request message so the check
+ # resolves to False (the stream is "connected").
+ async def _receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ req._receive = _receive
+ return req
+# Slice 5: payload size cap (413) + agent write rate limit (429).
+# ---------------------------------------------------------------------------
+
+
+class TestCanvasPayloadSizeCap:
+ """The 64 KB payload cap applies to all principals (agents and humans)."""
+
+ @pytest.mark.asyncio
+ async def test_create_oversized_payload_returns_413(self, ctx):
+ pid = await _new_project(ctx, "payload-cap")
+ oversized = "x" * (65 * 1024)
+ body = {
+ "kind": "text",
+ "x": 0, "y": 0, "w": 100, "h": 50,
+ "payload": {"text": oversized},
+ }
+ resp = await ctx.client.post(
+ f"/api/projects/{pid}/canvas/elements", json=body,
+ )
+ assert resp.status_code == 413, resp.text
+
+ @pytest.mark.asyncio
+ async def test_patch_oversized_payload_returns_413(self, ctx):
+ pid = await _new_project(ctx, "payload-cap-patch")
+ el = await _create_note(ctx.client, pid)
+ oversized = "x" * (65 * 1024)
+ resp = await ctx.client.patch(
+ f"/api/projects/{pid}/canvas/elements/{el['id']}",
+ json={"payload": {"text": oversized}},
+ )
+ assert resp.status_code == 413, resp.text
+
+ @pytest.mark.asyncio
+ async def test_create_normal_payload_succeeds(self, ctx):
+ pid = await _new_project(ctx, "payload-cap-ok")
+ body = {
+ "kind": "text",
+ "x": 0, "y": 0, "w": 100, "h": 50,
+ "payload": {"text": "normal content"},
+ }
+ resp = await ctx.client.post(
+ f"/api/projects/{pid}/canvas/elements", json=body,
+ )
+ assert resp.status_code == 201, resp.text
+
+
+class TestAgentWriteRateLimit:
+ """Agents are throttled to 30 canvas writes per 60 s rolling window.
+ Humans (session principals) are never throttled."""
+
+ @pytest.mark.asyncio
+ async def test_agent_exceeding_window_returns_429(self, ctx):
+ import time
+ import tinyagentos.routes.project_canvas as canvas_mod
+
+ pid = await _new_project(ctx, "rate-limit-agent")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, edit=True)
+
+ limiter = canvas_mod._canvas_write_limiter
+ max_attempts = canvas_mod._CANVAS_WRITE_MAX_ATTEMPTS
+ now = time.monotonic()
+ with limiter._lock:
+ limiter._log[cid] = [now - 1.0] * max_attempts
+
+ async with _bare(ctx.app) as bare:
+ resp = await bare.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "rate-limited"}},
+ headers=_hdr(token),
+ )
+ assert resp.status_code == 429, resp.text
+
+ @pytest.mark.asyncio
+ async def test_agent_delete_exceeding_window_returns_429(self, ctx):
+ import time
+ import tinyagentos.routes.project_canvas as canvas_mod
+
+ pid = await _new_project(ctx, "rate-limit-agent-del")
+ cid, token = await _mint_agent(ctx, pid, ("canvas_write",))
+ await _add_member(ctx, pid, cid)
+ await _grant_canvas(ctx, pid, cid, edit=True)
+
+ created = await ctx.client.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "to-delete"}},
+ )
+ assert created.status_code == 201
+ eid = created.json()["element"]["id"]
+
+ limiter = canvas_mod._canvas_write_limiter
+ max_attempts = canvas_mod._CANVAS_WRITE_MAX_ATTEMPTS
+ now = time.monotonic()
+ with limiter._lock:
+ limiter._log[cid] = [now - 1.0] * max_attempts
+
+ async with _bare(ctx.app) as bare:
+ resp = await bare.delete(
+ f"/api/projects/{pid}/canvas/elements/{eid}",
+ headers=_hdr(token),
+ )
+ assert resp.status_code == 429, resp.text
+
+ @pytest.mark.asyncio
+ async def test_human_write_not_rate_limited(self, ctx):
+ pid = await _new_project(ctx, "rate-limit-human")
+ for _ in range(31):
+ resp = await ctx.client.post(
+ f"/api/projects/{pid}/canvas/elements",
+ json={"kind": "note", "x": 0, "y": 0, "w": 1, "h": 1,
+ "payload": {"text": "human-write"}},
+ )
+ assert resp.status_code == 201, resp.text
+
diff --git a/tests/test_routes_projects.py b/tests/test_routes_projects.py
index aab5cdd3a..a1a0fd9a7 100644
--- a/tests/test_routes_projects.py
+++ b/tests/test_routes_projects.py
@@ -1,4 +1,5 @@
import pytest
+from httpx import ASGITransport, AsyncClient
@pytest.mark.asyncio
@@ -701,3 +702,78 @@ async def test_ready_tasks_filter_by_element(client):
)).json()["items"]
assert all(t["element_id"] is None for t in none_items)
assert len(none_items) == 1
+
+
+# ---------------------------------------------------------------------------
+# Slice 6 (D7): the Lead designation is an exclusive, project-level pointer
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_set_lead_exclusive_leaves_only_last(client):
+ """Setting lead B after lead A leaves only B; the pointer holds one lead."""
+ pid = (await client.post("/api/projects", json={"name": "P", "slug": "lead-excl"})).json()["id"]
+ await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": "agent-a"})
+ await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": "agent-b"})
+
+ r1 = await client.patch(f"/api/projects/{pid}/lead", json={"member_id": "agent-a"})
+ assert r1.status_code == 200
+ assert r1.json()["lead_member_id"] == "agent-a"
+
+ r2 = await client.patch(f"/api/projects/{pid}/lead", json={"member_id": "agent-b"})
+ assert r2.status_code == 200
+ assert r2.json()["lead_member_id"] == "agent-b"
+
+ # The pointer cannot hold two leads; only the last one remains.
+ proj = (await client.get(f"/api/projects/{pid}")).json()
+ assert proj["lead_member_id"] == "agent-b"
+
+
+@pytest.mark.asyncio
+async def test_set_lead_non_member_returns_404(client):
+ pid = (await client.post("/api/projects", json={"name": "P", "slug": "lead-404"})).json()["id"]
+ await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": "agent-a"})
+
+ r = await client.patch(f"/api/projects/{pid}/lead", json={"member_id": "not-a-member"})
+ assert r.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_clear_lead_to_null(client):
+ pid = (await client.post("/api/projects", json={"name": "P", "slug": "lead-null"})).json()["id"]
+ await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": "agent-a"})
+
+ r = await client.patch(f"/api/projects/{pid}/lead", json={"member_id": "agent-a"})
+ assert r.status_code == 200
+ assert r.json()["lead_member_id"] == "agent-a"
+ assert (await client.get(f"/api/projects/{pid}")).json()["lead_member_id"] == "agent-a"
+
+ r = await client.patch(f"/api/projects/{pid}/lead", json={"member_id": None})
+ assert r.status_code == 200
+ assert r.json()["lead_member_id"] is None
+ assert (await client.get(f"/api/projects/{pid}")).json()["lead_member_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_set_lead_requires_session(app, client):
+ """The lead route is session-only: an unauthenticated request is rejected."""
+ pid = (await client.post("/api/projects", json={"name": "P", "slug": "lead-auth"})).json()["id"]
+ await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": "agent-a"})
+
+ transport = ASGITransport(app=app)
+ async with AsyncClient(transport=transport, base_url="http://test") as anon:
+ r = await anon.patch(f"/api/projects/{pid}/lead", json={"member_id": "agent-a"})
+ assert r.status_code in (401, 403)
+
+
+@pytest.mark.asyncio
+async def test_old_per_member_lead_route_removed(client):
+ """The retired per-member lead route is gone (replaced by the project pointer)."""
+ pid = (await client.post("/api/projects", json={"name": "P", "slug": "lead-old"})).json()["id"]
+ await client.post(f"/api/projects/{pid}/members", json={"mode": "native", "agent_id": "agent-a"})
+
+ r = await client.patch(
+ f"/api/projects/{pid}/members/agent-a/lead", json={"is_lead": True}
+ )
+ assert r.status_code == 404
+
diff --git a/tinyagentos/agent_registry_store.py b/tinyagentos/agent_registry_store.py
index 48740bb88..ef4bfb9f5 100644
--- a/tinyagentos/agent_registry_store.py
+++ b/tinyagentos/agent_registry_store.py
@@ -49,6 +49,24 @@
);
"""
+# Partial unique index: at most ONE active agent may own any given non-empty
+# handle. SQLite enforces this atomically, so two concurrent consent approvals
+# of the same identity_claim cannot both flip to 'active' with the same handle.
+# Pending / suspended / revoked rows and empty handles are excluded from the
+# index, so (a) a handle can be reused once its owner leaves 'active', and (b)
+# pre-existing empty-handle active agents never block the index's creation.
+#
+# This index references the ``status`` column, which is added by
+# _migration_v1_add_status on the migration path. Per BaseStore's contract it
+# therefore CANNOT live in SCHEMA (the executescript runs before migrations and
+# would crash on a pre-status table); it is created in _post_init after the
+# migration guarantees the column exists.
+ACTIVE_HANDLE_INDEX = """
+CREATE UNIQUE INDEX IF NOT EXISTS ux_agent_active_handle
+ ON agent_registry(handle)
+ WHERE status = 'active' AND handle != '';
+"""
+
# ---------------------------------------------------------------------------
# Lifecycle state machine
# ---------------------------------------------------------------------------
@@ -348,6 +366,9 @@ async def _post_init(self) -> None:
await _migration_v1_add_status(self._db)
await _migration_v2_strip_at_display_name(self._db)
await _migration_v3_add_org_fields(self._db)
+ # Created after the status migration so the partial index's WHERE clause
+ # can reference the status column on the pre-status migration path.
+ await self._db.executescript(ACTIVE_HANDLE_INDEX)
# ------------------------------------------------------------------
# Registration
@@ -425,6 +446,26 @@ async def register(
# Read
# ------------------------------------------------------------------
+ async def rollback(self) -> None:
+ """Roll back the current transaction, clearing any failed write so the
+ connection can accept new statements. No-op if the store is closed or
+ no transaction is open.
+ """
+ if self._db is not None:
+ await self._db.rollback()
+
+ async def delete(self, canonical_id: str) -> None:
+ """Permanently remove *canonical_id* (used to clean up a half-registered
+ row when a concurrent approve wins the active-handle race). Returns
+ silently if *canonical_id* does not exist.
+ """
+ if self._db is None:
+ raise RuntimeError("AgentRegistryStore not initialised")
+ await self._db.execute(
+ "DELETE FROM agent_registry WHERE canonical_id = ?", (canonical_id,)
+ )
+ await self._db.commit()
+
async def get(self, canonical_id: str) -> Optional[dict]:
"""Return the record for *canonical_id*, or ``None``."""
if self._db is None:
diff --git a/tinyagentos/auth_middleware.py b/tinyagentos/auth_middleware.py
index 1a0e817ab..343c942f0 100644
--- a/tinyagentos/auth_middleware.py
+++ b/tinyagentos/auth_middleware.py
@@ -58,11 +58,29 @@
("POST", re.compile(rf"^/api/projects/{_SEG}/tasks/{_SEG}/(claim|release|close|reopen)$")),
)
+_AGENT_CANVAS_ROUTES = (
+ ("GET", re.compile(rf"^/api/projects/{_SEG}/canvas/elements$")),
+ ("POST", re.compile(rf"^/api/projects/{_SEG}/canvas/elements$")),
+ ("PATCH", re.compile(rf"^/api/projects/{_SEG}/canvas/elements/{_SEG}$")),
+ ("DELETE", re.compile(rf"^/api/projects/{_SEG}/canvas/elements/{_SEG}$")),
+ ("GET", re.compile(rf"^/api/projects/{_SEG}/canvas/snapshot\.png$")),
+ ("GET", re.compile(rf"^/api/projects/{_SEG}/canvas/snapshot\.tldr$")),
+ ("GET", re.compile(rf"^/api/projects/{_SEG}/canvas/stream$")),
+)
+
def _is_agent_task_path(method: str, path: str) -> bool:
"""True only for the exact subset of task routes a project_tasks token may
reach. Strict method + anchored-regex match; everything else is excluded."""
return any(m == method and rx.match(path) for m, rx in _AGENT_TASK_ROUTES)
+
+
+def _is_agent_canvas_path(method: str, path: str) -> bool:
+ """True only for the exact subset of canvas routes a canvas_read/canvas_write
+ token may reach. Strict method + anchored-regex match; the PATCH permissions
+ route stays session-only (owner/admin only) but the PATCH elements route is
+ reachable by a canvas_write-bound agent token, mirroring POST/DELETE."""
+ return any(m == method and rx.match(path) for m, rx in _AGENT_CANVAS_ROUTES)
# Bundle assets and the SPA shell HTML must be reachable without auth so:
# 1. The browser can install and cache the shell for offline / PWA use.
# 2. After a backend restart the cached shell loads immediately without
@@ -255,6 +273,7 @@ async def dispatch(self, request: Request, call_next):
if (
path in _AGENT_TOKEN_PATHS
or _is_agent_task_path(request.method, path)
+ or _is_agent_canvas_path(request.method, path)
) and auth_header.lower().startswith("bearer "):
request.state.user_id = None
request.state.is_admin = False
diff --git a/tinyagentos/projects/a2a.py b/tinyagentos/projects/a2a.py
index 5d944952b..85bdddcc1 100644
--- a/tinyagentos/projects/a2a.py
+++ b/tinyagentos/projects/a2a.py
@@ -104,31 +104,28 @@ def _resolve_member_names(member_rows: list[dict], config) -> set[str]:
return resolved
-def _resolve_lead_names(member_rows: list[dict], config) -> list[str]:
- """Return sorted list of resolved agent names for members where is_lead = 1.
-
- Uses the same config-lookup strategy as _resolve_member_names. Returns
- names in sorted order for deterministic storage.
+def _resolve_lead_names(project: dict, config) -> list[str]:
+ """Return the resolved agent name for the project's single exclusive lead.
+
+ The lead designation lives on the project as a single ``lead_member_id``
+ pointer (D7), not on the members. At most one name is ever returned; an
+ empty list means no lead. Uses the same config-lookup strategy as
+ _resolve_member_names so the quiet-filter and settings.leads stay in the
+ name-space the @mention parser and router expect.
"""
- leads = [m for m in member_rows if m.get("is_lead")]
- if not leads:
+ lead_id = project.get("lead_member_id")
+ if not lead_id:
return []
by_id, by_name = _build_agent_lookups(config)
- result: list[str] = []
- for m in leads:
- mid = m["member_id"]
- if by_id is None:
- # No config (test path): use member_id as name directly.
- result.append(mid)
- continue
- agent = by_id.get(mid) or by_name.get(mid)
- if agent is None:
- logger.debug("a2a: lead member_id %r not found in config — skipping", mid)
- continue
- name = agent.get("name")
- if name:
- result.append(name)
- return sorted(set(result))
+ if by_id is None:
+ # No config (test path): use the member id directly.
+ return [lead_id]
+ agent = by_id.get(lead_id) or by_name.get(lead_id)
+ if agent is None:
+ logger.debug("a2a: lead member_id %r not found in config — skipping", lead_id)
+ return []
+ name = agent.get("name")
+ return [name] if name else []
async def ensure_a2a_channel(
@@ -143,21 +140,21 @@ async def ensure_a2a_channel(
router both operate on names. When None (test path with name-keyed members),
member_ids are used as-is.
- Also syncs settings.leads from project members where is_lead = 1 so that
- the router can give leads visibility into all channel messages regardless
+ Also syncs settings.leads from the project's exclusive lead_member_id so that
+ the router can give the lead visibility into all channel messages regardless
of response_mode.
Idempotent. Serialized per project_id via _A2A_LOCKS to prevent racing
member-sync diffs when multiple callers fire concurrently.
"""
async with _A2A_LOCKS[project_id]:
+ project = await project_store.get_project(project_id)
project_members = await project_store.list_members(project_id)
expected = _resolve_member_names(project_members, config)
- expected_leads = _resolve_lead_names(project_members, config)
+ expected_leads = _resolve_lead_names(project, config)
matches = await _find_a2a_channels(channel_store, project_id)
if not matches:
- project = await project_store.get_project(project_id)
created_by = project.get("created_by", "system") if project else "system"
return await channel_store.create_channel(
name=A2A_NAME,
diff --git a/tinyagentos/projects/canvas/mcp_tools.py b/tinyagentos/projects/canvas/mcp_tools.py
index ce4cee46c..3d1c8812f 100644
--- a/tinyagentos/projects/canvas/mcp_tools.py
+++ b/tinyagentos/projects/canvas/mcp_tools.py
@@ -30,25 +30,66 @@ class CanvasToolContext:
data_root: Path
-async def canvas_list_elements(ctx: CanvasToolContext, *, project_id: str) -> dict:
+async def canvas_list_elements(
+ ctx: CanvasToolContext, *, project_id: str, agent_id: str
+) -> dict:
+ """List canvas elements an agent can read, gated by can_read_canvas (D3).
+
+ Surfaces the read-permission floor as a clean error, mirroring the other
+ in-process canvas tools so an agent without read access is told what to do
+ instead of silently receiving another principal's board contents."""
+ try:
+ await ctx.canvas_store.check_read_permission(
+ project_id, "agent", agent_id
+ )
+ except CanvasPermissionError:
+ return {
+ "error": "permission_denied",
+ "message": (
+ "This agent does not have read permission on the canvas. "
+ "Ask the user to enable it in project settings, or message "
+ "them to grant access."
+ ),
+ }
elements = await ctx.canvas_store.list_elements(project_id)
return {"elements": elements}
+async def _add_agent_element(
+ ctx: CanvasToolContext, *, project_id: str, agent_id: str, element: dict
+) -> dict:
+ """Add an element authored by an agent, surfacing the edit-permission gate
+ as a clean error (mirrors canvas_update_element / canvas_delete_element)."""
+ try:
+ el = await ctx.canvas_store.add_element(
+ project_id=project_id,
+ author_kind="agent", author_id=agent_id,
+ element=element,
+ )
+ except CanvasPermissionError:
+ return {
+ "error": "permission_denied",
+ "message": (
+ "This agent does not have edit permission on the canvas. "
+ "Ask the user to enable it in project settings, or message "
+ "them to make the change."
+ ),
+ }
+ return {"element": el}
+
+
async def canvas_add_note(
ctx: CanvasToolContext, *, project_id: str, agent_id: str,
text: str, x: float, y: float, color: str = "yellow",
) -> dict:
- el = await ctx.canvas_store.add_element(
- project_id=project_id,
- author_kind="agent", author_id=agent_id,
+ return await _add_agent_element(
+ ctx, project_id=project_id, agent_id=agent_id,
element={
"kind": "note", "x": float(x), "y": float(y),
"w": 200.0, "h": 100.0,
"payload": {"text": text, "color": color, "font_size": 14},
},
)
- return {"element": el}
async def canvas_add_link(
@@ -56,77 +97,67 @@ async def canvas_add_link(
url: str, x: float, y: float,
) -> dict:
meta = await fetch_link_metadata(url)
- el = await ctx.canvas_store.add_element(
- project_id=project_id,
- author_kind="agent", author_id=agent_id,
+ return await _add_agent_element(
+ ctx, project_id=project_id, agent_id=agent_id,
element={
"kind": "link", "x": float(x), "y": float(y),
"w": 320.0, "h": 120.0, "payload": meta,
},
)
- return {"element": el}
async def canvas_add_image(
ctx: CanvasToolContext, *, project_id: str, agent_id: str,
file_id: str, x: float, y: float, alt: str = "",
) -> dict:
- el = await ctx.canvas_store.add_element(
- project_id=project_id,
- author_kind="agent", author_id=agent_id,
+ return await _add_agent_element(
+ ctx, project_id=project_id, agent_id=agent_id,
element={
"kind": "image", "x": float(x), "y": float(y),
"w": 240.0, "h": 240.0,
"payload": {"file_id": file_id, "alt": alt, "mime": "image/png"},
},
)
- return {"element": el}
async def canvas_add_text(
ctx: CanvasToolContext, *, project_id: str, agent_id: str,
text: str, x: float, y: float, font_size: int = 16,
) -> dict:
- el = await ctx.canvas_store.add_element(
- project_id=project_id,
- author_kind="agent", author_id=agent_id,
+ return await _add_agent_element(
+ ctx, project_id=project_id, agent_id=agent_id,
element={
"kind": "text", "x": float(x), "y": float(y),
"w": 220.0, "h": 80.0,
"payload": {"text": text, "font_size": int(font_size)},
},
)
- return {"element": el}
async def canvas_add_mermaid(
ctx: CanvasToolContext, *, project_id: str, agent_id: str,
source: str, x: float, y: float,
) -> dict:
- el = await ctx.canvas_store.add_element(
- project_id=project_id,
- author_kind="agent", author_id=agent_id,
+ return await _add_agent_element(
+ ctx, project_id=project_id, agent_id=agent_id,
element={
"kind": "mermaid", "x": float(x), "y": float(y),
"w": 320.0, "h": 240.0, "payload": {"source": source},
},
)
- return {"element": el}
async def canvas_add_flowchart(
ctx: CanvasToolContext, *, project_id: str, agent_id: str,
source: str, x: float, y: float,
) -> dict:
- el = await ctx.canvas_store.add_element(
- project_id=project_id,
- author_kind="agent", author_id=agent_id,
+ return await _add_agent_element(
+ ctx, project_id=project_id, agent_id=agent_id,
element={
"kind": "flowchart", "x": float(x), "y": float(y),
"w": 320.0, "h": 240.0, "payload": {"source": source},
},
)
- return {"element": el}
async def canvas_add_mindmap_edge(
@@ -147,9 +178,8 @@ async def canvas_add_mindmap_edge(
}
ax, ay = a["x"] + a["w"] / 2, a["y"] + a["h"] / 2
bx, by = b["x"] + b["w"] / 2, b["y"] + b["h"] / 2
- el = await ctx.canvas_store.add_element(
- project_id=project_id,
- author_kind="agent", author_id=agent_id,
+ return await _add_agent_element(
+ ctx, project_id=project_id, agent_id=agent_id,
element={
"kind": "mindmap_edge",
"x": min(ax, bx), "y": min(ay, by),
@@ -157,7 +187,6 @@ async def canvas_add_mindmap_edge(
"payload": {"from": from_id, "to": to_id},
},
)
- return {"element": el}
async def canvas_update_element(
diff --git a/tinyagentos/projects/canvas/store.py b/tinyagentos/projects/canvas/store.py
index 55995e66a..c627574f7 100644
--- a/tinyagentos/projects/canvas/store.py
+++ b/tinyagentos/projects/canvas/store.py
@@ -124,6 +124,10 @@ async def add_element(
raise ValueError(f"agents may not emit kind={kind}")
if author_kind not in ("user", "agent"):
raise ValueError(f"invalid author_kind: {author_kind}")
+ # Enforce the per-project edit permission for agents (defense in depth:
+ # the route already gates on scope + can_edit_canvas, but the store is
+ # the authoritative floor so a direct caller can never bypass it).
+ await self._check_edit_permission(project_id, author_kind, author_id)
eid = element.get("id") or new_id("cve")
now = time.time()
# Upsert: the client may re-send an element it already created (e.g. a
@@ -154,7 +158,10 @@ async def add_element(
)
await self._db.commit()
new_el = await self.get_element(eid)
- await self._publish(project_id, "canvas.element_added", {"element": new_el})
+ await self._publish(
+ project_id, "canvas.element_added",
+ {"element": new_el, "actor": {"kind": author_kind, "id": author_id}},
+ )
return new_el
async def list_elements(
@@ -203,6 +210,30 @@ async def _check_edit_permission(
f"agent {author_id} has no can_edit_canvas on project {project_id}"
)
+ async def check_read_permission(
+ self, project_id: str, author_kind: str, author_id: str
+ ) -> None:
+ """Gate canvas reads on can_read_canvas for agents (D3 read matrix).
+
+ Mirrors _check_edit_permission: humans bypass the floor, agents must hold
+ the per-project can_read_canvas flag, and a missing member row fails
+ closed. This is the in-process analogue of the route-level canvas_read
+ scope check, guarding the agent MCP read path at the store floor."""
+ if author_kind == "user":
+ return
+ if author_kind != "agent":
+ raise ValueError(f"invalid author_kind: {author_kind}")
+ async with self._db.execute(
+ "SELECT can_read_canvas FROM project_members "
+ "WHERE project_id = ? AND member_id = ?",
+ (project_id, author_id),
+ ) as cur:
+ row = await cur.fetchone()
+ if row is None or not row[0]:
+ raise CanvasPermissionError(
+ f"agent {author_id} has no can_read_canvas on project {project_id}"
+ )
+
async def update_element(
self,
*,
@@ -239,7 +270,10 @@ async def update_element(
updated = await self.get_element(element_id, project_id=project_id)
if updated is None:
raise ValueError(f"element not found: {element_id}")
- await self._publish(project_id, "canvas.element_updated", {"element": updated})
+ await self._publish(
+ project_id, "canvas.element_updated",
+ {"element": updated, "actor": {"kind": author_kind, "id": author_id}},
+ )
return updated
async def delete_element(
@@ -262,5 +296,5 @@ async def delete_element(
if cur.rowcount == 1:
await self._publish(
project_id, "canvas.element_deleted",
- {"element_id": element_id},
+ {"element_id": element_id, "actor": {"kind": author_kind, "id": author_id}},
)
diff --git a/tinyagentos/projects/project_store.py b/tinyagentos/projects/project_store.py
index 875b75fd2..ecabdbeb7 100644
--- a/tinyagentos/projects/project_store.py
+++ b/tinyagentos/projects/project_store.py
@@ -20,6 +20,7 @@
updated_at REAL NOT NULL,
archived_at REAL,
deleted_at REAL,
+ lead_member_id TEXT,
settings TEXT NOT NULL DEFAULT '{}'
);
CREATE INDEX IF NOT EXISTS idx_projects_status ON projects(status);
@@ -67,8 +68,10 @@ class ProjectStore(BaseStore):
async def _post_init(self) -> None:
for col_def in (
"ALTER TABLE project_members ADD COLUMN can_edit_canvas INTEGER NOT NULL DEFAULT 0",
+ "ALTER TABLE project_members ADD COLUMN can_read_canvas INTEGER NOT NULL DEFAULT 0",
"ALTER TABLE project_members ADD COLUMN is_lead INTEGER NOT NULL DEFAULT 0",
"ALTER TABLE projects ADD COLUMN user_id TEXT NOT NULL DEFAULT ''",
+ "ALTER TABLE projects ADD COLUMN lead_member_id TEXT",
):
try:
await self._db.execute(col_def)
@@ -76,16 +79,50 @@ async def _post_init(self) -> None:
except Exception:
# Column already exists on fresh installs (created by SCHEMA).
pass
+ # D7 backfill: the exclusive per-project lead moves from the per-member
+ # is_lead flag (non-exclusive) to the project's single lead_member_id
+ # pointer. Only projects with EXACTLY one flagged member are migrated;
+ # any with zero or several flagged members are left NULL so a human picks
+ # in the UI (the old flag never promised exclusivity).
+ await self._backfill_lead_member_id()
- async def set_member_lead(self, project_id: str, member_id: str, is_lead: bool) -> None:
- val = 1 if is_lead else 0
- cur = await self._db.execute(
- "UPDATE project_members SET is_lead = ? WHERE project_id = ? AND member_id = ?",
- (val, project_id, member_id),
+ async def _backfill_lead_member_id(self) -> None:
+ rows = await (await self._db.execute(
+ "SELECT id FROM projects WHERE lead_member_id IS NULL"
+ )).fetchall()
+ for (pid,) in rows:
+ cur = await self._db.execute(
+ "SELECT member_id FROM project_members WHERE project_id = ? AND is_lead = 1",
+ (pid,),
+ )
+ flagged = [r[0] for r in await cur.fetchall()]
+ if len(flagged) == 1:
+ await self._db.execute(
+ "UPDATE projects SET lead_member_id = ? WHERE id = ?",
+ (flagged[0], pid),
+ )
+ await self._db.commit()
+
+ async def set_lead(self, project_id: str, member_id: "str | None") -> None:
+ """Set (or clear, when member_id is None) the project's exclusive lead.
+
+ The single pointer column makes the one-lead-per-project invariant
+ structural: setting a new lead atomically unsets any previous one, and
+ there is no partial-update window. A member_id that is not a current
+ member of the project raises KeyError (the route maps it to 404).
+ """
+ p = await self.get_project(project_id)
+ if p is None:
+ raise KeyError(f"project {project_id!r} not found")
+ if member_id is not None:
+ member = await self.get_member(project_id, member_id)
+ if member is None:
+ raise KeyError(f"member {member_id!r} not in project {project_id!r}")
+ await self._db.execute(
+ "UPDATE projects SET lead_member_id = ? WHERE id = ?",
+ (member_id, project_id),
)
await self._db.commit()
- if cur.rowcount == 0:
- raise KeyError(f"member {member_id!r} not in project {project_id!r}")
async def create_project(
self,
@@ -233,6 +270,13 @@ async def remove_member(self, project_id: str, member_id: str) -> None:
"DELETE FROM project_members WHERE project_id = ? AND member_id = ?",
(project_id, member_id),
)
+ # Removing the designated lead unsets the pointer so it can never dangle
+ # on a member that no longer belongs to the project.
+ await self._db.execute(
+ "UPDATE projects SET lead_member_id = NULL "
+ "WHERE id = ? AND lead_member_id = ?",
+ (project_id, member_id),
+ )
await self._db.commit()
async def list_members(self, project_id: str) -> list[dict]:
@@ -244,6 +288,17 @@ async def list_members(self, project_id: str) -> list[dict]:
keys = [d[0] for d in cur.description]
return [dict(zip(keys, r)) for r in rows]
+ async def get_member(self, project_id: str, member_id: str) -> dict | None:
+ async with self._db.execute(
+ "SELECT * FROM project_members WHERE project_id = ? AND member_id = ?",
+ (project_id, member_id),
+ ) as cur:
+ row = await cur.fetchone()
+ if row is None:
+ return None
+ keys = [d[0] for d in cur.description]
+ return dict(zip(keys, row))
+
async def log_activity(
self,
project_id: str,
diff --git a/tinyagentos/routes/agent_auth_requests.py b/tinyagentos/routes/agent_auth_requests.py
index 4e4349e3e..afc5e18c6 100644
--- a/tinyagentos/routes/agent_auth_requests.py
+++ b/tinyagentos/routes/agent_auth_requests.py
@@ -28,7 +28,8 @@
from fastapi.responses import JSONResponse
from pydantic import BaseModel
-from tinyagentos.agent_registry_store import mint_registry_token
+from aiosqlite import IntegrityError
+from tinyagentos.agent_registry_store import _slugify, mint_registry_token
from tinyagentos.auth_context import CurrentUser, current_user
logger = logging.getLogger(__name__)
@@ -54,6 +55,12 @@
# agent's OWN project only (bound by the token's project_id claim). Does NOT
# grant task create, member management, or project lifecycle.
"project_tasks",
+ # Canvas access: read and write on a specific project's canvas. Like
+ # project_tasks, a project_id is required so the token is bound to the
+ # operator-validated project rather than whatever the unauthenticated agent
+ # named in the request.
+ "canvas_read",
+ "canvas_write",
})
@@ -298,17 +305,26 @@ async def _do_approve(request: Request, request_id: str, body: ApproveBody, user
),
)
- # project_tasks binds the token to a specific project and adds a membership
- # row, so require the human to pick that project explicitly in the consent
- # card. Never fall back to the agent-supplied project_id for a task grant:
- # POST /api/agents/auth-requests is unauthenticated, so the request could
- # name any existing project the operator never validated. Other scopes keep
- # the fallback so global tokens still work. Checked before any registration
- # so a rejected approval never leaves an orphaned agent.
- if "project_tasks" in body.granted_scopes and body.project_id is None:
+ _CANVAS_SCOPES = {"canvas_read", "canvas_write"}
+ _PROJECT_SCOPES = {"project_tasks"} | _CANVAS_SCOPES
+
+ # project_tasks and the canvas scopes bind the token to a specific project
+ # and add a membership row, so require the human to pick that project
+ # explicitly in the consent card. Never fall back to the agent-supplied
+ # project_id for these grants: POST /api/agents/auth-requests is
+ # unauthenticated, so the request could name any existing project the
+ # operator never validated. Other scopes keep the fallback so global
+ # tokens still work. Checked before any registration so a rejected approval
+ # never leaves an orphaned agent.
+ needs_project = bool(set(body.granted_scopes) & _PROJECT_SCOPES)
+ # Reject None, "", and whitespace-only: a blank project_id is not a real
+ # binding, and a downstream truthy check would treat it as unbound, so an
+ # empty string must fail closed exactly like a missing one.
+ if needs_project and not (body.project_id and body.project_id.strip()):
+ missing = sorted(set(body.granted_scopes) & _PROJECT_SCOPES)
raise HTTPException(
status_code=400,
- detail="project_id is required when granting project_tasks",
+ detail=f"project_id is required when granting {missing}",
)
registry = _get_registry_store(request)
@@ -324,20 +340,62 @@ async def _do_approve(request: Request, request_id: str, body: ApproveBody, user
# non-empty value.
_claim = record["identity_claim"].strip().removeprefix("@").strip()
display_name = _claim or record["framework"]
- reg_record = await registry.register(
- framework=record["framework"],
- display_name=display_name,
- user_id=user.user_id,
- origin="external-selfjoin",
- handle="",
- )
- canonical_id = reg_record["canonical_id"]
- # Consent approval IS the activation. external-selfjoin agents are born
- # 'pending' (governance lifecycle); approving the auth-request transitions
- # them to 'active' so they are NOT in the bus inactive/revocation feed and
- # @taOSmd's identity-AND-grant gate accepts them.
- await registry.set_status(canonical_id, "active", actor=user.user_id)
+ handle = _slugify(_claim)
+ existing_active = await registry.get_by_handle(handle, status="active")
+ if existing_active is not None:
+ raise HTTPException(
+ status_code=409,
+ detail=(
+ f"handle '{handle}' is already in use by active agent "
+ f"{existing_active['canonical_id']}; pick a different identity_claim"
+ ),
+ )
+
+ # Register with the handle SET at birth. external-selfjoin agents are born
+ # 'pending' (governance lifecycle), so the partial unique index
+ # ux_agent_active_handle (active + non-empty handle) does not fire at INSERT
+ # time; it fires only when we flip the row to 'active' below. That removes
+ # the old window (register handle='' -> set active -> update handle) in which
+ # an agent could sit ACTIVE with an empty handle, and lets SQLite reject a
+ # duplicate active handle the instant a concurrent approve tries to take it.
+ canonical_id = None
+ try:
+ reg_record = await registry.register(
+ framework=record["framework"],
+ display_name=display_name,
+ user_id=user.user_id,
+ origin="external-selfjoin",
+ handle=handle,
+ )
+ canonical_id = reg_record["canonical_id"]
+
+ # Consent approval IS the activation. external-selfjoin agents are born
+ # 'pending' (governance lifecycle); approving the auth-request transitions
+ # them to 'active' so they are NOT in the bus inactive/revocation feed and
+ # @taOSmd's identity-AND-grant gate accepts them.
+ await registry.set_status(canonical_id, "active", actor=user.user_id)
+ except IntegrityError:
+ # A concurrent approve already took this active handle (the partial
+ # unique index fired). Roll back the failed write and remove the
+ # half-registered pending row so we never leave an active-without-handle
+ # agent or a stale pending row. Return the same friendly 409.
+ try:
+ await registry.rollback()
+ except Exception: # noqa: BLE001 - never mask the 409 below
+ pass
+ if canonical_id is not None:
+ try:
+ await registry.delete(canonical_id)
+ except Exception: # noqa: BLE001 - never mask the 409 below
+ pass
+ raise HTTPException(
+ status_code=409,
+ detail=(
+ f"handle '{handle}' is already in use by active agent; "
+ f"pick a different identity_claim"
+ ),
+ )
# Resolve effective project binding: admin override wins; fall back to the
# project_id the agent requested (may be None for a global token).
diff --git a/tinyagentos/routes/agent_registry.py b/tinyagentos/routes/agent_registry.py
index 59597a484..e5c8e4576 100644
--- a/tinyagentos/routes/agent_registry.py
+++ b/tinyagentos/routes/agent_registry.py
@@ -92,6 +92,7 @@ class OrgUpdateRequest(BaseModel):
"files_read", "files_write",
"tools_execute", "registry_feeds_read",
"project_tasks",
+ "canvas_read", "canvas_write",
})
diff --git a/tinyagentos/routes/project_canvas.py b/tinyagentos/routes/project_canvas.py
index 99c72de91..17153d523 100644
--- a/tinyagentos/routes/project_canvas.py
+++ b/tinyagentos/routes/project_canvas.py
@@ -7,9 +7,12 @@
import asyncio
import json
import logging
+import threading
+import time
+from collections import OrderedDict
from typing import Literal
-from fastapi import APIRouter, Request
+from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse, FileResponse, Response
from pydantic import BaseModel, Field
@@ -20,14 +23,137 @@
logger = logging.getLogger(__name__)
router = APIRouter()
+_CANVAS_READ_SCOPE = "canvas_read"
+_CANVAS_WRITE_SCOPE = "canvas_write"
+
+_CANVAS_PAYLOAD_MAX_BYTES = 64 * 1024
+_CANVAS_WRITE_MAX_ATTEMPTS = 30
+_CANVAS_WRITE_WINDOW_SECONDS = 60
+_CANVAS_WRITE_MAX_KEYS = 10_000
+
+
+class _CanvasWriteLimiter:
+ def __init__(self, max_attempts: int = _CANVAS_WRITE_MAX_ATTEMPTS,
+ window_seconds: int = _CANVAS_WRITE_WINDOW_SECONDS):
+ self._max = max_attempts
+ self._window = window_seconds
+ self._log: OrderedDict[str, list[float]] = OrderedDict()
+ self._lock = threading.Lock()
+
+ def _prune(self, key: str) -> None:
+ cutoff = time.monotonic() - self._window
+ if key not in self._log:
+ return
+ self._log[key] = [t for t in self._log[key] if t > cutoff]
+ if not self._log[key]:
+ del self._log[key]
+ else:
+ self._log.move_to_end(key)
+
+ def _ensure_capacity(self) -> None:
+ while len(self._log) >= _CANVAS_WRITE_MAX_KEYS:
+ self._log.popitem(last=False)
+
+ def is_limited(self, key: str) -> bool:
+ with self._lock:
+ self._prune(key)
+ return len(self._log.get(key, [])) >= self._max
+
+ def record(self, key: str) -> None:
+ with self._lock:
+ self._prune(key)
+ if key not in self._log:
+ self._ensure_capacity()
+ self._log[key] = []
+ self._log[key].append(time.monotonic())
+ self._log.move_to_end(key)
+
+
+_canvas_write_limiter = _CanvasWriteLimiter()
+
+
+def _check_payload_size(payload: dict) -> None:
+ size = len(json.dumps(payload, default=str).encode("utf-8"))
+ if size > _CANVAS_PAYLOAD_MAX_BYTES:
+ raise HTTPException(
+ status_code=413,
+ detail=f"canvas payload exceeds {_CANVAS_PAYLOAD_MAX_BYTES // 1024} KB limit",
+ )
+
def _user_id(request: Request) -> str:
- user = getattr(request.state, "user", None)
- if user and isinstance(user, dict) and "id" in user:
- return user["id"]
+ uid = getattr(request.state, "user_id", None)
+ if uid:
+ return uid
return "system"
+async def _authorize_canvas_actor(
+ request: Request, project_id: str, mode: Literal["read", "write"]
+) -> "tuple[str, str] | JSONResponse":
+ """Resolve + authorize the actor for a canvas route.
+
+ Accepts EITHER a session owner/admin (behavior unchanged from before the
+ agent gate) OR an approved external agent's registry JWT bound to THIS
+ project with the matching canvas scope AND the matching per-project member
+ flag:
+
+ * read mode -> canvas_read scope + can_read_canvas member flag
+ * write mode -> canvas_write scope + can_edit_canvas member flag
+
+ Returns ``(actor_kind, actor_id)`` on success, or a JSONResponse to return
+ directly. A token bound to a DIFFERENT project collapses into an
+ existence-hiding 404 (never confirms the project exists). A token for this
+ project that is missing the scope or the member flag gets 403.
+ """
+ uid = getattr(request.state, "user_id", None)
+ if uid:
+ # Session path: project visibility gate (D3). Only the project owner or
+ # an admin may touch the canvas; a human needs no scope and no checkbox,
+ # but a non-owner collapses into the SAME existence-hiding 404 the agent
+ # path uses (compare _get_owned_project on the task routes). Attribution
+ # stays the verified user.
+ ps = request.app.state.project_store
+ project = await ps.get_project(project_id)
+ is_admin = bool(getattr(request.state, "is_admin", False))
+ if project is not None and not is_admin and project.get("user_id") != uid:
+ return JSONResponse({"error": "not found"}, status_code=404)
+ return ("user", _user_id(request))
+ auth_header = request.headers.get("Authorization", "")
+ if not auth_header.lower().startswith("bearer "):
+ # Middleware normally 401s unauthenticated requests before the route
+ # runs; a middleware-bypassing test context reaches here, so fall back
+ # to a system actor (there is no real principal to attribute to).
+ return ("user", "system")
+ from tinyagentos.agent_token_auth import (
+ check_agent_scope_for_project,
+ PROJECT_SCOPE_MISMATCH_DETAIL,
+ )
+ scope = _CANVAS_READ_SCOPE if mode == "read" else _CANVAS_WRITE_SCOPE
+ try:
+ cid = await check_agent_scope_for_project(request, scope, project_id)
+ except HTTPException as exc:
+ if exc.status_code == 403 and exc.detail == PROJECT_SCOPE_MISMATCH_DETAIL:
+ return JSONResponse({"error": "not found"}, status_code=404)
+ raise
+ if cid is None:
+ return JSONResponse({"error": "not found"}, status_code=404)
+ ps = request.app.state.project_store
+ member = await ps.get_member(project_id, cid)
+ flag = (member or {}).get(
+ "can_read_canvas" if mode == "read" else "can_edit_canvas"
+ )
+ if not flag:
+ return JSONResponse(
+ {
+ "error": "permission_denied",
+ "message": f"agent {cid} lacks canvas {mode} access on {project_id}",
+ },
+ status_code=403,
+ )
+ return ("agent", cid)
+
+
class CreateElementIn(BaseModel):
kind: Literal[
"note", "link", "image", "user_shape",
@@ -48,6 +174,9 @@ class CreateElementIn(BaseModel):
async def list_canvas_elements(
project_id: str, request: Request, element_id: str | None = None,
):
+ auth = await _authorize_canvas_actor(request, project_id, "read")
+ if isinstance(auth, JSONResponse):
+ return auth
cs = request.app.state.project_canvas_store
elements = await cs.list_elements(project_id, element_id=element_id)
return {"elements": elements}
@@ -57,8 +186,18 @@ async def list_canvas_elements(
async def create_canvas_element(
project_id: str, payload: CreateElementIn, request: Request,
):
- cs = request.app.state.project_canvas_store
+ auth = await _authorize_canvas_actor(request, project_id, "write")
+ if isinstance(auth, JSONResponse):
+ return auth
+ actor_kind, actor_id = auth
+ if actor_kind == "agent" and _canvas_write_limiter.is_limited(actor_id):
+ return JSONResponse(
+ {"error": "rate_limit", "message": "too many canvas writes, try again later"},
+ status_code=429,
+ )
element = payload.model_dump()
+ _check_payload_size(element.get("payload") or {})
+ cs = request.app.state.project_canvas_store
element_id = element.pop("element_id", None)
if element["kind"] == "link":
url = (element.get("payload") or {}).get("url")
@@ -69,11 +208,13 @@ async def create_canvas_element(
try:
new_el = await cs.add_element(
project_id=project_id, element=element,
- author_kind="user", author_id=_user_id(request),
+ author_kind=actor_kind, author_id=actor_id,
element_id=element_id,
)
except ValueError as e:
return JSONResponse({"error": str(e)}, status_code=400)
+ if actor_kind == "agent":
+ _canvas_write_limiter.record(actor_id)
return {"element": new_el}
@@ -91,39 +232,67 @@ class PatchElementIn(BaseModel):
async def update_canvas_element(
project_id: str, element_id: str, payload: PatchElementIn, request: Request,
):
+ auth = await _authorize_canvas_actor(request, project_id, "write")
+ if isinstance(auth, JSONResponse):
+ return auth
+ actor_kind, actor_id = auth
+ if actor_kind == "agent" and _canvas_write_limiter.is_limited(actor_id):
+ return JSONResponse(
+ {"error": "rate_limit", "message": "too many canvas writes, try again later"},
+ status_code=429,
+ )
cs = request.app.state.project_canvas_store
patch = {k: v for k, v in payload.model_dump().items() if v is not None}
+ if "payload" in patch:
+ _check_payload_size(patch["payload"])
try:
updated = await cs.update_element(
project_id=project_id, element_id=element_id, patch=patch,
- author_kind="user", author_id=_user_id(request),
+ author_kind=actor_kind, author_id=actor_id,
)
except CanvasPermissionError as e:
return JSONResponse({"error": "permission_denied", "message": str(e)}, status_code=403)
except ValueError as e:
return JSONResponse({"error": str(e)}, status_code=404)
+ if actor_kind == "agent":
+ _canvas_write_limiter.record(actor_id)
return {"element": updated}
@router.delete("/api/projects/{project_id}/canvas/elements/{element_id}", status_code=204)
async def delete_canvas_element(project_id: str, element_id: str, request: Request):
+ auth = await _authorize_canvas_actor(request, project_id, "write")
+ if isinstance(auth, JSONResponse):
+ return auth
+ actor_kind, actor_id = auth
+ if actor_kind == "agent" and _canvas_write_limiter.is_limited(actor_id):
+ return JSONResponse(
+ {"error": "rate_limit", "message": "too many canvas writes, try again later"},
+ status_code=429,
+ )
cs = request.app.state.project_canvas_store
try:
await cs.delete_element(
project_id=project_id, element_id=element_id,
- author_kind="user", author_id=_user_id(request),
+ author_kind=actor_kind, author_id=actor_id,
)
except CanvasPermissionError as e:
return JSONResponse({"error": "permission_denied", "message": str(e)}, status_code=403)
+ if actor_kind == "agent":
+ _canvas_write_limiter.record(actor_id)
return Response(status_code=204)
class PermissionIn(BaseModel):
- can_edit_canvas: bool
+ can_read_canvas: bool | None = None
+ can_edit_canvas: bool | None = None
@router.get("/api/projects/{project_id}/canvas/snapshot.png")
async def get_canvas_png(project_id: str, request: Request):
+ auth = await _authorize_canvas_actor(request, project_id, "read")
+ if isinstance(auth, JSONResponse):
+ return auth
cs = request.app.state.project_canvas_store
elements = await cs.list_elements(project_id)
project = await request.app.state.project_store.get_project(project_id)
@@ -141,6 +310,9 @@ async def get_canvas_png(project_id: str, request: Request):
@router.get("/api/projects/{project_id}/canvas/snapshot.tldr")
async def get_canvas_tldr(project_id: str, request: Request):
+ auth = await _authorize_canvas_actor(request, project_id, "read")
+ if isinstance(auth, JSONResponse):
+ return auth
snap = request.app.state.canvas_snapshotter
path = await snap.export_now(project_id)
if path is None or not path.exists():
@@ -153,29 +325,68 @@ async def set_canvas_permission(
project_id: str, agent_id: str, payload: PermissionIn, request: Request,
):
ps = request.app.state.project_store
- val = 1 if payload.can_edit_canvas else 0
+ project = await ps.get_project(project_id)
+ if project is None:
+ return JSONResponse({"error": "not found"}, status_code=404)
+ uid = getattr(request.state, "user_id", None)
+ is_admin = bool(getattr(request.state, "is_admin", False))
+ if not uid or (not is_admin and project.get("user_id") != uid):
+ return JSONResponse(
+ {
+ "error": "forbidden",
+ "message": "only the project owner or an admin may change canvas permissions",
+ },
+ status_code=403,
+ )
+ sets: list[str] = []
+ params: list = []
+ if payload.can_read_canvas is not None:
+ sets.append("can_read_canvas = ?")
+ params.append(1 if payload.can_read_canvas else 0)
+ if payload.can_edit_canvas is not None:
+ sets.append("can_edit_canvas = ?")
+ params.append(1 if payload.can_edit_canvas else 0)
+ if not sets:
+ return JSONResponse({"error": "no permission field provided"}, status_code=400)
+ params.extend([project_id, agent_id])
cur = await ps._db.execute(
- "UPDATE project_members SET can_edit_canvas = ? "
+ f"UPDATE project_members SET {', '.join(sets)} "
"WHERE project_id = ? AND member_id = ?",
- (val, project_id, agent_id),
+ params,
)
await ps._db.commit()
if cur.rowcount == 0:
return JSONResponse({"error": "member not found"}, status_code=404)
+ member = await ps.get_member(project_id, agent_id)
broker = request.app.state.project_event_broker
from tinyagentos.projects.events import ProjectEvent
await broker.publish(
project_id,
ProjectEvent(
kind="canvas.permission_changed",
- payload={"agent_id": agent_id, "can_edit_canvas": bool(val)},
+ payload={
+ "actor": {"kind": "user", "id": uid},
+ "agent_id": agent_id,
+ "can_read_canvas": bool(member.get("can_read_canvas")),
+ "can_edit_canvas": bool(member.get("can_edit_canvas")),
+ },
),
)
- return {"ok": True, "agent_id": agent_id, "can_edit_canvas": bool(val)}
+ return {
+ "ok": True,
+ "agent_id": agent_id,
+ "can_read_canvas": bool(member.get("can_read_canvas")),
+ "can_edit_canvas": bool(member.get("can_edit_canvas")),
+ }
@router.get("/api/projects/{project_id}/canvas/stream")
async def canvas_stream(project_id: str, request: Request):
+ auth = await _authorize_canvas_actor(request, project_id, "read")
+ if isinstance(auth, JSONResponse):
+ return auth
+ actor_kind, actor_id = auth
+ ps = request.app.state.project_store
broker = request.app.state.project_event_broker
queue = await broker.subscribe(project_id)
@@ -187,6 +398,13 @@ async def gen():
try:
ev = await asyncio.wait_for(queue.get(), timeout=10.0)
except asyncio.TimeoutError:
+ # Keepalive tick: re-verify a live agent principal's read
+ # access so a revoked/removed agent cannot keep a long-lived
+ # SSE open (slice 4). Session users are unaffected.
+ if actor_kind == "agent":
+ member = await ps.get_member(project_id, actor_id)
+ if not (member or {}).get("can_read_canvas"):
+ return
yield ":keepalive\n\n"
continue
if not str(ev.kind).startswith("canvas."):
diff --git a/tinyagentos/routes/projects.py b/tinyagentos/routes/projects.py
index c7a6d4b14..e2bd8d5e9 100644
--- a/tinyagentos/routes/projects.py
+++ b/tinyagentos/routes/projects.py
@@ -296,27 +296,38 @@ async def list_members(
return {"items": await store.list_members(project_id)}
-class LeadIn(BaseModel):
- is_lead: bool
+class ProjectLeadIn(BaseModel):
+ member_id: "str | None" = None
-@router.patch("/api/projects/{project_id}/members/{member_id}/lead")
-async def set_lead(
+@router.patch("/api/projects/{project_id}/lead")
+async def set_project_lead(
project_id: str,
- member_id: str,
- body: LeadIn,
+ body: ProjectLeadIn,
request: Request,
user: CurrentUser = Depends(current_user),
):
+ """Set the project's exclusive Lead (D7). The single lead_member_id pointer
+ makes the one-lead-per-project invariant structural: setting a new lead
+ atomically unsets the previous one. ``member_id: null`` clears the lead.
+
+ Session-only (owner or admin, same gate as the members routes). A member id
+ not in the project returns 404.
+ """
store = request.app.state.project_store
p = await store.get_project(project_id)
if p is None:
return JSONResponse({"error": "not found"}, status_code=404)
require_owner_or_admin(user, p["user_id"])
try:
- await store.set_member_lead(project_id, member_id, body.is_lead)
+ await store.set_lead(project_id, body.member_id)
except KeyError as e:
return JSONResponse({"error": str(e)}, status_code=404)
+ await store.log_activity(
+ project_id, user.user_id, "project.lead_changed", {"member_id": body.member_id}
+ )
+ members = await store.list_members(project_id)
+ _mirror(request, {**p, "members": members})
try:
from tinyagentos.projects.a2a import ensure_a2a_channel
await ensure_a2a_channel(
@@ -327,7 +338,7 @@ async def set_lead(
)
except Exception:
logger.warning("a2a ensure failed for project %s on set_lead", project_id, exc_info=True)
- return {"ok": True, "is_lead": body.is_lead}
+ return {"ok": True, "lead_member_id": body.member_id}
@router.delete("/api/projects/{project_id}/members/{member_id}")