From dc2caa1b5a8ec632e16e6dae122e85eff49a9e65 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Sun, 19 Jul 2026 14:51:54 +0200 Subject: [PATCH 1/3] feat(collab): agent delegation handshake + sponsor_contact_id + cascades (D1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement cross-user agent delegation (milestone D1 of #2012): - Add sponsor_contact_id column to agent_registry (schema + migration + store API) - Delegation-request envelope dispatch in peer inbox (kind=delegation_request) - Policy gate with per-project auto_approve_delegation knob (default OFF) - Sponsored scope tiers: {a2a_send, a2a_receive, project_tasks, canvas_read, registry_feeds_read} - Hard-deny files_write and decisions_write at scope validation - Decisions card integration: collab_delegation_gate kind triggers invite mint on approve - Revoke cascade: contact/membership change → revoke sponsored identities → unassign tasks - 3-level kill-switch: per-agent (existing), per-contact pause, per-instance panic - Invite metadata column with sponsor_contact_id propagation through redeem flow - Project store helpers: is_project_member(), get/set_project_setting() Fixes: #2019 --- tests/test_collab_d1_delegation.py | 300 ++++++++++++ tinyagentos/agent_registry_store.py | 104 ++++- tinyagentos/delegation_handler.py | 537 ++++++++++++++++++++++ tinyagentos/projects/invite_store.py | 27 +- tinyagentos/projects/project_store.py | 60 ++- tinyagentos/routes/agent_auth_requests.py | 10 + tinyagentos/routes/decisions.py | 62 ++- tinyagentos/routes/peer.py | 19 +- tinyagentos/routes/project_invites.py | 20 + 9 files changed, 1108 insertions(+), 31 deletions(-) create mode 100644 tests/test_collab_d1_delegation.py create mode 100644 tinyagentos/delegation_handler.py diff --git a/tests/test_collab_d1_delegation.py b/tests/test_collab_d1_delegation.py new file mode 100644 index 000000000..eab6bd0ed --- /dev/null +++ b/tests/test_collab_d1_delegation.py @@ -0,0 +1,300 @@ +"""Tests for cross-user collab D1 — agent delegation handshake + sponsor_contact_id.""" + +from __future__ import annotations + +import json +import time + +import pytest + + +# --------------------------------------------------------------------------- +# Scope denylist tests +# --------------------------------------------------------------------------- + +class TestDelegationScopeValidation: + def test_hard_denies_files_write(self): + from tinyagentos.delegation_handler import validate_delegation_scopes + + granted, denied = validate_delegation_scopes( + ["a2a_send", "files_write", "project_tasks"] + ) + assert "files_write" in denied + assert "files_write" not in granted + assert "a2a_send" in granted + assert "project_tasks" in granted + + def test_hard_denies_decisions_write(self): + from tinyagentos.delegation_handler import validate_delegation_scopes + + granted, denied = validate_delegation_scopes( + ["decisions_write", "canvas_read"] + ) + assert "decisions_write" in denied + assert "decisions_write" not in granted + assert "canvas_read" in granted + + def test_allows_default_scopes(self): + from tinyagentos.delegation_handler import validate_delegation_scopes + from tinyagentos.delegation_handler import SPONSORED_DEFAULT_SCOPES + + granted, denied = validate_delegation_scopes(list(SPONSORED_DEFAULT_SCOPES)) + assert len(denied) == 0 + assert set(granted) == SPONSORED_DEFAULT_SCOPES + + def test_empty_request_returns_no_scopes(self): + from tinyagentos.delegation_handler import validate_delegation_scopes + + granted, denied = validate_delegation_scopes([]) + assert granted == [] + assert denied == [] + + +# --------------------------------------------------------------------------- +# Envelope body validation tests +# --------------------------------------------------------------------------- + +class TestDelegationEnvelopeValidation: + def test_valid_envelope_body(self): + from tinyagentos.delegation_handler import _validate_delegation_envelope_body + + ok, err, parsed = _validate_delegation_envelope_body({ + "agent_slug": "grok-taos", + "display_name": "Grok TAOS", + "requested_scopes": ["a2a_send", "project_tasks"], + "project_id": "prj-123", + }) + assert ok is True + assert err == "" + assert parsed is not None + assert parsed["agent_slug"] == "grok-taos" + assert parsed["display_name"] == "Grok TAOS" + assert parsed["project_id"] == "prj-123" + + def test_missing_field(self): + from tinyagentos.delegation_handler import _validate_delegation_envelope_body + + ok, err, parsed = _validate_delegation_envelope_body({ + "agent_slug": "grok-taos", + "display_name": "Grok TAOS", + }) + assert ok is False + assert "missing required field" in err + assert parsed is None + + def test_empty_scopes(self): + from tinyagentos.delegation_handler import _validate_delegation_envelope_body + + ok, err, parsed = _validate_delegation_envelope_body({ + "agent_slug": "grok-taos", + "display_name": "Grok TAOS", + "requested_scopes": [], + "project_id": "prj-123", + }) + assert ok is False + assert "must not be empty" in err + + def test_scopes_not_a_list(self): + from tinyagentos.delegation_handler import _validate_delegation_envelope_body + + ok, err, parsed = _validate_delegation_envelope_body({ + "agent_slug": "grok-taos", + "display_name": "Grok TAOS", + "requested_scopes": "not-a-list", + "project_id": "prj-123", + }) + assert ok is False + assert "must be a list" in err + + def test_empty_agent_slug(self): + from tinyagentos.delegation_handler import _validate_delegation_envelope_body + + ok, err, parsed = _validate_delegation_envelope_body({ + "agent_slug": "", + "display_name": "Grok TAOS", + "requested_scopes": ["a2a_send"], + "project_id": "prj-123", + }) + assert ok is False + assert "must be a non-empty string" in err + + +# --------------------------------------------------------------------------- +# Sponsor list / set tests +# --------------------------------------------------------------------------- + +class TestSponsorRegistryMethods: + @pytest.mark.asyncio + async def test_list_by_sponsor_empty(self, tmp_path): + from tinyagentos.agent_registry_store import AgentRegistryStore + + db_path = tmp_path / "test_registry.db" + store = AgentRegistryStore(db_path) + await store.init() + + sponsored = await store.list_by_sponsor("hub:hogne") + assert sponsored == [] + await store.close() + + @pytest.mark.asyncio + async def test_list_by_sponsor_with_registration(self, tmp_path): + from tinyagentos.agent_registry_store import AgentRegistryStore + + db_path = tmp_path / "test_registry.db" + store = AgentRegistryStore(db_path) + await store.init() + + # Register with sponsor + reg = await store.register( + framework="test", + display_name="Sponsored Agent", + user_id="user-1", + origin="external-selfjoin", + handle="sponsored-agent", + sponsor_contact_id="hub:hogne", + ) + canonical_id = reg["canonical_id"] + # List by sponsor + sponsored = await store.list_by_sponsor("hub:hogne") + assert len(sponsored) == 1 + assert sponsored[0]["sponsor_contact_id"] == "hub:hogne" + + # List by different sponsor — empty + other = await store.list_by_sponsor("hub:other") + assert other == [] + await store.close() + + @pytest.mark.asyncio + async def test_list_by_sponsor_filter_by_status(self, tmp_path): + from tinyagentos.agent_registry_store import AgentRegistryStore + + db_path = tmp_path / "test_registry.db" + store = AgentRegistryStore(db_path) + await store.init() + + await store.register( + framework="test", + display_name="Active Sponsored", + user_id="user-1", + origin="external-selfjoin", + handle="active-sponsor", + sponsor_contact_id="hub:hogne", + ) + # Only active agents + active = await store.list_by_sponsor("hub:hogne", status="active") + assert len(active) == 0 # external-selfjoin starts pending + + # With no status filter, shows all + all_sponsored = await store.list_by_sponsor("hub:hogne") + assert len(all_sponsored) == 1 # pending agent + await store.close() + + @pytest.mark.asyncio + async def test_set_sponsor(self, tmp_path): + from tinyagentos.agent_registry_store import AgentRegistryStore + + db_path = tmp_path / "test_registry.db" + store = AgentRegistryStore(db_path) + await store.init() + + reg = await store.register( + framework="test", + display_name="Agent", + user_id="user-1", + origin="taos-deployed", + handle="test-agent", + ) + cid = reg["canonical_id"] + + # Initially NULL + agent = await store.get(cid) + assert agent["sponsor_contact_id"] is None + + # Set sponsor + await store.set_sponsor(cid, "hub:hogne") + agent = await store.get(cid) + assert agent["sponsor_contact_id"] == "hub:hogne" + + # Clear sponsor + await store.set_sponsor(cid, None) + agent = await store.get(cid) + assert agent["sponsor_contact_id"] is None + await store.close() + + @pytest.mark.asyncio + async def test_migration_adds_sponsor_column(self, tmp_path): + from tinyagentos.agent_registry_store import _migration_v5_add_sponsor_contact_id + + import aiosqlite + + db_path = tmp_path / "test_registry.db" + # Simulate pre-migration DB with agent_registry but no sponsor_contact_id + conn = await aiosqlite.connect(db_path) + await conn.execute(""" + CREATE TABLE agent_registry ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + canonical_id TEXT NOT NULL UNIQUE, + display_name TEXT NOT NULL DEFAULT '', + framework TEXT NOT NULL DEFAULT '', + user_id TEXT NOT NULL DEFAULT '', + origin TEXT NOT NULL DEFAULT 'taos-deployed', + handle TEXT NOT NULL DEFAULT '', + role TEXT, + capabilities TEXT NOT NULL DEFAULT '[]', + created_ts TEXT NOT NULL, + revoked_at TEXT, + status TEXT NOT NULL DEFAULT 'active' + ) + """) + await conn.commit() + + # Migration should add the column + await _migration_v5_add_sponsor_contact_id(conn) + + # Verify column exists + cols = {row[1] for row in await (await conn.execute( + "PRAGMA table_info(agent_registry)" + )).fetchall()} + assert "sponsor_contact_id" in cols + + # Idempotent + await _migration_v5_add_sponsor_contact_id(conn) + + await conn.close() + + +# --------------------------------------------------------------------------- +# Invite metadata tests +# --------------------------------------------------------------------------- + +class TestInviteMetadata: + @pytest.mark.asyncio + async def test_row_to_dict_deserializes_metadata(self, tmp_path): + from tinyagentos.projects.invite_store import ProjectInviteStore + + db_path = tmp_path / "test_invites.db" + store = ProjectInviteStore(db_path) + await store.init() + + metadata = { + "kind": "delegation_sponsored", + "sponsor_contact_id": "hub:hogne", + "agent_slug": "grok-taos", + } + + result = await store.mint( + project_id="prj-test", + scopes=["project_tasks", "a2a_send"], + approval_mode="auto", + check_interval_secs=1800, + created_by="hub:hogne", + metadata=metadata, + ) + + invite_id = result["record"]["invite_id"] + invite = await store.get(invite_id) + # metadata should be deserialized to a dict + assert isinstance(invite["metadata"], dict) + assert invite["metadata"]["kind"] == "delegation_sponsored" + assert invite["metadata"]["sponsor_contact_id"] == "hub:hogne" + await store.close() diff --git a/tinyagentos/agent_registry_store.py b/tinyagentos/agent_registry_store.py index 96f837337..fdbe17949 100644 --- a/tinyagentos/agent_registry_store.py +++ b/tinyagentos/agent_registry_store.py @@ -34,18 +34,19 @@ SCHEMA = """ CREATE TABLE IF NOT EXISTS agent_registry ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - canonical_id TEXT NOT NULL UNIQUE, - display_name TEXT NOT NULL DEFAULT '', - framework TEXT NOT NULL DEFAULT '', - user_id TEXT NOT NULL DEFAULT '', - origin TEXT NOT NULL DEFAULT 'taos-deployed', - handle TEXT NOT NULL DEFAULT '', - role TEXT, - capabilities TEXT NOT NULL DEFAULT '[]', - created_ts TEXT NOT NULL, - revoked_at TEXT, - status TEXT NOT NULL DEFAULT 'active' + id INTEGER PRIMARY KEY AUTOINCREMENT, + canonical_id TEXT NOT NULL UNIQUE, + display_name TEXT NOT NULL DEFAULT '', + framework TEXT NOT NULL DEFAULT '', + user_id TEXT NOT NULL DEFAULT '', + origin TEXT NOT NULL DEFAULT 'taos-deployed', + handle TEXT NOT NULL DEFAULT '', + role TEXT, + capabilities TEXT NOT NULL DEFAULT '[]', + created_ts TEXT NOT NULL, + revoked_at TEXT, + status TEXT NOT NULL DEFAULT 'active', + sponsor_contact_id TEXT ); """ @@ -205,6 +206,26 @@ async def _migration_v4_dedupe_active_handles(conn) -> None: await conn.commit() +async def _migration_v5_add_sponsor_contact_id(conn) -> None: + """Add sponsor_contact_id column (idempotent) for Cross-User Collab D1. + + Delegated agents (those minted via a contact's delegation-request flow) + carry ``sponsor_contact_id`` linking back to the sponsoring contact. + Existing rows (non-delegated) get NULL. + """ + existing_cols = { + row[1] + for row in await ( + await conn.execute("PRAGMA table_info(agent_registry)") + ).fetchall() + } + if "sponsor_contact_id" not in existing_cols: + await conn.execute( + "ALTER TABLE agent_registry ADD COLUMN sponsor_contact_id TEXT" + ) + await conn.commit() + + # --------------------------------------------------------------------------- # Signing-key helpers (Ed25519, persisted to disk) # --------------------------------------------------------------------------- @@ -417,6 +438,7 @@ async def _post_init(self) -> None: # Dedupe BEFORE the index so a pre-invariant DB with duplicate active # handles cannot make the CREATE UNIQUE INDEX (hence boot) fail. await _migration_v4_dedupe_active_handles(self._db) + await _migration_v5_add_sponsor_contact_id(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. # Guard the index creation too: if some path we did not anticipate still @@ -447,6 +469,7 @@ async def register( title: Optional[str] = None, reports_to: Optional[str] = None, capabilities: Optional[list[str]] = None, + sponsor_contact_id: Optional[str] = None, ) -> dict: """Mint a canonical_id, persist the record, and return it. @@ -454,6 +477,10 @@ async def register( not exist yet, so it cannot be part of an existing cycle) - use ``set_reporting`` after registration to validate a manager change. + ``sponsor_contact_id`` is set for delegated agents (cross-user collab + D1) — the contact_id of the human who sponsored this agent. NULL for + all other origins. + Raises ``RuntimeError`` if the store is not initialised. """ if self._db is None: @@ -487,11 +514,13 @@ async def register( """ INSERT INTO agent_registry (canonical_id, display_name, framework, user_id, origin, - handle, role, title, reports_to, capabilities, created_ts, status) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + handle, role, title, reports_to, capabilities, created_ts, + status, sponsor_contact_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (canonical_id, display_name, framework, user_id, origin, - handle, role, title, reports_to, caps_json, created_ts, initial_status), + handle, role, title, reports_to, caps_json, created_ts, + initial_status, sponsor_contact_id), ) await self._db.commit() @@ -615,6 +644,51 @@ async def list_inactive(self) -> list[dict]: rows = await cursor.fetchall() return [{"canonical_id": r["canonical_id"], "status": r["status"]} for r in rows] + async def list_by_sponsor( + self, sponsor_contact_id: str, *, status: Optional[str] = None + ) -> list[dict]: + """Return all registry records sponsored by *sponsor_contact_id*. + + Used for cascade revocation: when a contact or their project membership + is revoked, all sponsored agent identities are revoked and their tokens + invalidated. Optionally filtered by *status*. + """ + if self._db is None: + raise RuntimeError("AgentRegistryStore not initialised") + if status is not None: + cursor = await self._db.execute( + "SELECT * FROM agent_registry " + "WHERE sponsor_contact_id = ? AND status = ? ORDER BY id", + (sponsor_contact_id, status), + ) + else: + cursor = await self._db.execute( + "SELECT * FROM agent_registry " + "WHERE sponsor_contact_id = ? ORDER BY id", + (sponsor_contact_id,), + ) + rows = await cursor.fetchall() + return [_row_to_dict(r) for r in rows] + + async def set_sponsor( + self, canonical_id: str, sponsor_contact_id: Optional[str] + ) -> Optional[dict]: + """Set (or, with ``None``, clear) the sponsor_contact_id on a registry row. + + Returns the updated record, or None if *canonical_id* does not exist. + """ + if self._db is None: + raise RuntimeError("AgentRegistryStore not initialised") + record = await self.get(canonical_id) + if record is None: + return None + await self._db.execute( + "UPDATE agent_registry SET sponsor_contact_id = ? WHERE canonical_id = ?", + (sponsor_contact_id, canonical_id), + ) + await self._db.commit() + return await self.get(canonical_id) + # ------------------------------------------------------------------ # Lifecycle state machine # ------------------------------------------------------------------ diff --git a/tinyagentos/delegation_handler.py b/tinyagentos/delegation_handler.py new file mode 100644 index 000000000..072777757 --- /dev/null +++ b/tinyagentos/delegation_handler.py @@ -0,0 +1,537 @@ +"""Agent delegation handler — cross-user collab D1. + +Processes delegation-request envelopes from remote contacts, applies +scope denylist and policy gates, creates Decisions cards for manual +approval, and on approval mints project invites for the sponsored agent. + +Also provides cascade revocation and kill-switch machinery. +""" + +from __future__ import annotations + +import json +import logging +import time +from typing import Optional + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Scope denylist — hard-coded scopes that can NEVER be granted to sponsored +# agents in v1, regardless of owner decisions (design section 5). +# --------------------------------------------------------------------------- + +SPONSORED_DENY_SCOPES: frozenset[str] = frozenset({ + "files_write", + "decisions_write", +}) + +# Default scope tier for delegated agents per design section 5. +SPONSORED_DEFAULT_SCOPES: frozenset[str] = frozenset({ + "a2a_send", + "a2a_receive", + "project_tasks", + "canvas_read", + "registry_feeds_read", +}) + +# Scopes that bind the JWT to a specific project (same set as agent_auth_requests). +_PROJECT_SCOPES: frozenset[str] = frozenset({"project_tasks", "canvas_read", "canvas_write"}) + + +def validate_delegation_scopes( + requested_scopes: list[str], +) -> tuple[list[str], list[str]]: + """Validate and filter scopes for a delegation request. + + Returns ``(granted_scopes, denied_scopes)``. Scopes in + ``SPONSORED_DENY_SCOPES`` are stripped with a warning; scopes outside + ``SPONSORED_DEFAULT_SCOPES`` require explicit per-scope Decisions approval + but are NOT auto-denied — they surface to the human for approval. + + The returned ``granted_scopes`` are safe-to-mint; any scope in + ``denied_scopes`` was hard-denied and will never be included in the minted + invite. + """ + requested_set = set(requested_scopes) + denied = sorted(requested_set & SPONSORED_DENY_SCOPES) + allowed = sorted(requested_set - SPONSORED_DENY_SCOPES) + if denied: + logger.warning( + "delegation: hard-denied scopes %r from request %r", + denied, sorted(requested_set), + ) + return allowed, denied + + +def _validate_delegation_envelope_body(body: dict) -> tuple[bool, str, Optional[dict]]: + """Validate the body of a delegation-request envelope. + + Returns ``(ok, error, parsed)``. ``parsed`` is a dict with keys + ``agent_slug``, ``display_name``, ``requested_scopes``, and ``project_id`` + when valid. + """ + required = ("agent_slug", "display_name", "requested_scopes", "project_id") + for field in required: + if field not in body: + return False, f"missing required field: {field}", None + value = body[field] + if field == "requested_scopes": + if not isinstance(value, list): + return False, f"{field} must be a list", None + if not value: + return False, f"{field} must not be empty", None + elif not isinstance(value, str) or not value.strip(): + return False, f"{field} must be a non-empty string", None + + requested_scopes = body["requested_scopes"] + unknown = sorted(set(requested_scopes) - SPONSORED_DEFAULT_SCOPES - SPONSORED_DENY_SCOPES) + # Unknown scopes are not immediately denied — they require explicit + # per-scope Decisions approval. We just log them here; the decision + # created later will surface them to the human. + if unknown: + logger.info( + "delegation: elevated scopes in request: %r (require explicit approval)", + unknown, + ) + + parsed = { + "agent_slug": body["agent_slug"].strip(), + "display_name": body["display_name"].strip(), + "requested_scopes": body["requested_scopes"], + "project_id": body["project_id"].strip(), + } + return True, "", parsed + + +async def process_delegation_request( + request, + *, + contact_id: str, + envelope_body: dict, +) -> dict: + """Process a delegation-request envelope from a remote contact. + + Called from the peer inbox when envelope.kind == "delegation_request". + + 1. Validate envelope body structure. + 2. Verify the contact has ``member_kind="human"`` in the target project. + 3. Apply scope denylist (strip ``files_write``, ``decisions_write``). + 4. Check ``auto_approve_delegation`` project setting. + - If ON (dev-swarm future): auto-mint invite, return result. + - If OFF (v1 default): create a blocking Decisions card. + + Returns a dict suitable for the peer inbox response envelope. + On pending-approval: ``{"status": "pending_approval", "decision_id": ...}`` + On auto-approved: ``{"status": "approved", "invite_id": ..., ...}`` + """ + ok, err, parsed = _validate_delegation_envelope_body(envelope_body) + if not ok or parsed is None: + return {"status": "error", "error": err} + + agent_slug = parsed["agent_slug"] + display_name = parsed["display_name"] + requested_scopes = parsed["requested_scopes"] + project_id = parsed["project_id"] + + # Verify sender is an active human collaborator on the target project. + project_store = getattr(request.app.state, "project_store", None) + if project_store is None: + return {"status": "error", "error": "project store not available"} + + if not await project_store.is_project_member( + project_id, contact_id, member_kind="human" + ): + return { + "status": "error", + "error": ( + f"contact {contact_id} is not a human collaborator on project " + f"{project_id}" + ), + } + + # Apply scope denylist. + granted_scopes, denied_scopes = validate_delegation_scopes(requested_scopes) + + # Check project policy: auto_approve_delegation knob. + auto_approve = await project_store.get_project_setting( + project_id, "auto_approve_delegation", default=False + ) + + if auto_approve: + # Future dev-swarm path: immediate approval. + return await _auto_approve_delegation( + request, + contact_id=contact_id, + agent_slug=agent_slug, + display_name=display_name, + granted_scopes=granted_scopes, + denied_scopes=denied_scopes, + project_id=project_id, + ) + + # v1 manual path: create a blocking Decisions card. + return await _create_delegation_decision( + request, + contact_id=contact_id, + agent_slug=agent_slug, + display_name=display_name, + granted_scopes=granted_scopes, + denied_scopes=denied_scopes, + project_id=project_id, + ) + + +async def _create_delegation_decision( + request, + *, + contact_id: str, + agent_slug: str, + display_name: str, + granted_scopes: list[str], + denied_scopes: list[str], + project_id: str, +) -> dict: + """Create a blocking Decisions card for manual delegation approval. + + The human sees: "contact {contact_id} wants to delegate agent + '{display_name}' ({agent_slug}) to project {project_id} with scopes + {granted_scopes}." Any hard-denied scopes are noted in the question + text so the human knows they were stripped. + """ + decision_store = getattr(request.app.state, "decision_store", None) + if decision_store is None: + return {"status": "error", "error": "decision store not available"} + + question_parts = [ + f"{contact_id} wants to delegate agent " + f"'{display_name}' ({agent_slug}) to this project", + f"Requested scopes: {', '.join(granted_scopes)}", + ] + if denied_scopes: + question_parts.append( + f"(Hard-denied: {', '.join(denied_scopes)} — " + f"these scopes cannot be granted to delegated agents in v1)" + ) + + question = ". ".join(question_parts) + "." + + try: + decision = await decision_store.create( + from_agent=contact_id, + question=question, + type="approve_deny", + priority="blocking", + project_id=project_id, + metadata={ + "kind": "collab_delegation_gate", + "contact_id": contact_id, + "agent_slug": agent_slug, + "display_name": display_name, + "granted_scopes": granted_scopes, + "denied_scopes": denied_scopes, + "project_id": project_id, + }, + ) + except Exception: + logger.warning( + "delegation: failed to create decision for %s / %s", + contact_id, agent_slug, exc_info=True, + ) + return {"status": "error", "error": "failed to create approval decision"} + + return { + "status": "pending_approval", + "decision_id": decision["id"], + } + + +async def _auto_approve_delegation( + request, + *, + contact_id: str, + agent_slug: str, + display_name: str, + granted_scopes: list[str], + denied_scopes: list[str], + project_id: str, +) -> dict: + """Auto-approve a delegation request (dev-swarm future path). + + Mints a project invite (kind="agent", pin_required=false) and returns + the invite_id + connection_bundle. + """ + invite_store = getattr(request.app.state, "project_invite_store", None) + if invite_store is None: + return {"status": "error", "error": "invite store not available"} + + # The project-scoped scopes require a non-null project_id for the JWT. + project_scopes = sorted(set(granted_scopes) & _PROJECT_SCOPES) + non_project_scopes = sorted(set(granted_scopes) - _PROJECT_SCOPES) + + try: + invite = await invite_store.mint( + project_id=project_id, + scopes=non_project_scopes + project_scopes, + approval_mode="auto", + check_interval_secs=1800, + created_by=contact_id, + metadata={ + "kind": "delegation_sponsored", + "sponsor_contact_id": contact_id, + "agent_slug": agent_slug, + "display_name": display_name, + "pin_required": False, + }, + ) + except Exception: + logger.warning( + "delegation: failed to create invite for %s / %s", + contact_id, agent_slug, exc_info=True, + ) + return {"status": "error", "error": "failed to create project invite"} + + return { + "status": "approved", + "invite_id": invite["id"], + "agent_slug": agent_slug, + } + + +async def complete_delegation_approval( + request, + *, + decision_metadata: dict, +) -> dict: + """Complete a delegation approval after the human answers the Decisions card. + + Called from the decisions route when a delegation_gate decision is approved. + ``decision_metadata`` is the metadata dict stored on the decision at creation + time (contains contact_id, agent_slug, display_name, granted_scopes, etc.). + + Mints a project invite and returns the result for delivery over the peer + channel. + """ + contact_id = decision_metadata.get("contact_id", "") + agent_slug = decision_metadata.get("agent_slug", "") + display_name = decision_metadata.get("display_name", "") + granted_scopes = decision_metadata.get("granted_scopes", []) + project_id = decision_metadata.get("project_id", "") + + if not all([contact_id, agent_slug, display_name, project_id]): + return {"status": "error", "error": "incomplete decision metadata"} + + invite_store = getattr(request.app.state, "project_invite_store", None) + if invite_store is None: + return {"status": "error", "error": "invite store not available"} + + try: + invite = await invite_store.mint( + project_id=project_id, + scopes=granted_scopes, + approval_mode="auto", + check_interval_secs=1800, + created_by=contact_id, + metadata={ + "kind": "delegation_sponsored", + "sponsor_contact_id": contact_id, + "agent_slug": agent_slug, + "display_name": display_name, + "pin_required": False, + }, + ) + except Exception: + logger.warning( + "delegation: complete_approval: failed to create invite for %s / %s", + contact_id, agent_slug, exc_info=True, + ) + return {"status": "error", "error": "failed to create project invite"} + + return { + "status": "approved", + "invite_id": invite["id"], + "agent_slug": agent_slug, + } + + +async def cascade_sponsor_revoke( + request, + *, + contact_id: str, + project_id: str | None = None, + reason: str = "sponsor revoked", +) -> dict: + """Revoke all sponsored agent identities for a contact. + + When *project_id* is provided, only revoke tokens bound to that project + (membership-revoke). When None, revoke ALL sponsored identities + (contact-revoke). + + Also unassigns in-flight board tasks and posts an A2A system line. + Returns a summary dict of what was revoked. + """ + registry = getattr(request.app.state, "agent_registry", None) + if registry is None: + return {"status": "error", "error": "registry not available"} + + # Find all active agents sponsored by this contact. + sponsored = await registry.list_by_sponsor(contact_id, status="active") + revoked_ids: list[str] = [] + + for agent in sponsored: + canonical_id = agent["canonical_id"] + try: + await registry.set_status(canonical_id, "revoked", actor=contact_id) + revoked_ids.append(canonical_id) + except Exception: + logger.warning( + "cascade_revoke: failed to revoke %s (sponsor %s)", + canonical_id, contact_id, exc_info=True, + ) + + # Unassign in-flight board tasks for all revoked agent identities. + task_store = getattr(request.app.state, "project_task_store", None) + unassigned_count = 0 + if task_store is not None and revoked_ids: + for canonical_id in revoked_ids: + try: + # Find tasks assigned to this agent and move them back to ready. + unassigned = await _unassign_agent_tasks( + task_store, canonical_id, project_id + ) + unassigned_count += unassigned + except Exception: + logger.warning( + "cascade_revoke: task unassign failed for %s", + canonical_id, exc_info=True, + ) + + # Post A2A system line. + try: + from tinyagentos.projects.a2a import ensure_a2a_channel + + if project_id: + channel = await ensure_a2a_channel( + request.app.state.chat_channels, + request.app.state.project_store, + project_id, + config=getattr(request.app.state, "config", None), + ) + msg_store = request.app.state.chat_messages + await msg_store.send_message( + channel_id=channel["id"], + author_id="system", + author_type="user", + content=( + f"Collaboration with {contact_id} has been revoked. " + f"{len(revoked_ids)} delegated agent(s) revoked, " + f"{unassigned_count} task(s) unassigned." + ), + ) + except Exception: + logger.warning( + "cascade_revoke: A2A system line failed for %s", + contact_id, exc_info=True, + ) + + return { + "status": "revoked", + "contact_id": contact_id, + "revoked_agents": len(revoked_ids), + "unassigned_tasks": unassigned_count, + "revoked_ids": revoked_ids, + "reason": reason, + } + + +async def _unassign_agent_tasks( + task_store, + canonical_id: str, + project_id: str | None = None, +) -> int: + """Unassign all in-flight tasks for *canonical_id*, moving them back to 'ready'. + + Returns the count of tasks unassigned. + """ + # The task store interface varies by implementation; use the available + # method to find and unassign. We query for tasks with assignee matching + # the agent's canonical_id and reset them. + try: + tasks = await task_store.list_for_assignee(canonical_id) + except AttributeError: + # Fall back: try list_tasks with filter. + try: + all_tasks = await task_store.list_tasks( + project_id=project_id, + status="in_progress", + ) + tasks = [t for t in all_tasks if t.get("assignee_id") == canonical_id] + except Exception: + return 0 + + count = 0 + for task in tasks: + task_id = task.get("id") or task.get("task_id") + if not task_id: + continue + try: + await task_store.update_task(task_id, assignee_id=None, status="ready") + count += 1 + except Exception: + pass + + return count + + +# --------------------------------------------------------------------------- +# Kill-switch: 3 levels per design section 5 +# --------------------------------------------------------------------------- + +async def kill_switch_per_contact(request, *, contact_id: str) -> dict: + """Level 2 kill-switch: pause collaboration with a specific contact. + + Suspends the peer link and revokes all sponsored tokens. Reversible + (the peer link can be re-established). + """ + contacts_store = getattr(request.app.state, "contacts_store", None) + if contacts_store is None: + return {"status": "error", "error": "contacts store not available"} + + # Suspend the peer link (revoke the inbound token, set status to suspended). + try: + await contacts_store.revoke_peer_link(contact_id) + except Exception: + logger.warning( + "kill_switch_per_contact: peer link revoke failed for %s", + contact_id, exc_info=True, + ) + + # Cascade to all sponsored agents. + cascade_result = await cascade_sponsor_revoke( + request, contact_id=contact_id, reason="kill-switch (per-contact pause)", + ) + + return { + "status": "paused", + "contact_id": contact_id, + "revoked_agents": cascade_result.get("revoked_agents", 0), + "unassigned_tasks": cascade_result.get("unassigned_tasks", 0), + } + + +async def kill_switch_per_instance(request) -> dict: + """Level 3 kill-switch: disable the entire peer route family. + + Sets a flag on app.state that the peer routes check on every request. + Does NOT revoke tokens — just blocks all new peer traffic. + Reversible by clearing the flag. + """ + # Set a flag that the peer routes check. + request.app.state._peer_disabled = True + logger.warning("kill_switch: per-instance panic — peer routes DISABLED") + return {"status": "panic", "peer_routes": "disabled"} + + +def is_peer_disabled(request) -> bool: + """Check whether the per-instance peer panic switch is active.""" + return getattr(request.app.state, "_peer_disabled", False) diff --git a/tinyagentos/projects/invite_store.py b/tinyagentos/projects/invite_store.py index 0ef9ab208..a88da801e 100644 --- a/tinyagentos/projects/invite_store.py +++ b/tinyagentos/projects/invite_store.py @@ -29,7 +29,8 @@ status TEXT NOT NULL, redeemed_by TEXT, redeemed_request_id TEXT, - display_name TEXT + display_name TEXT, + metadata TEXT NOT NULL DEFAULT '{}' ); """ @@ -89,6 +90,11 @@ async def _post_init(self) -> None: "ALTER TABLE project_invites ADD COLUMN display_name TEXT" ) await self._db.commit() + if "metadata" not in existing_cols: + await self._db.execute( + "ALTER TABLE project_invites ADD COLUMN metadata TEXT NOT NULL DEFAULT '{}'" + ) + await self._db.commit() await self._db.execute( "CREATE INDEX IF NOT EXISTS idx_project_invites_project " "ON project_invites(project_id)" @@ -107,7 +113,8 @@ def _generate_pin(self) -> str: async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, check_interval_secs: int, created_by: str, - display_name: str | None = None) -> dict: + display_name: str | None = None, + metadata: dict | None = None) -> dict: if self._db is None: raise RuntimeError("ProjectInviteStore not initialised") @@ -158,8 +165,8 @@ async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, INSERT INTO project_invites (invite_id, project_id, pin_hash, scopes, approval_mode, check_interval_secs, created_by, created_ts, expires_ts, status, - display_name) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?) + display_name, metadata) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?) """, ( invite_id, @@ -172,6 +179,7 @@ async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, now, expires_ts, display_name, + json.dumps(metadata or {}), ), ) await self._db.commit() @@ -204,6 +212,7 @@ async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, "redeemed_by": None, "redeemed_request_id": None, "display_name": display_name, + "metadata": metadata or {}, }, "pin": pin, } @@ -361,4 +370,12 @@ async def _fetch_row(self, invite_id: str) -> aiosqlite.Row | None: def _row_to_dict(self, row: aiosqlite.Row) -> dict: if row is None: return {} - return dict(row) + d = dict(row) + # Deserialise JSON columns. + for field in ("scopes", "metadata"): + if field in d and isinstance(d[field], str): + try: + d[field] = json.loads(d[field]) + except (ValueError, TypeError): + pass + return d diff --git a/tinyagentos/projects/project_store.py b/tinyagentos/projects/project_store.py index d6b006b1e..1bb9fe8a5 100644 --- a/tinyagentos/projects/project_store.py +++ b/tinyagentos/projects/project_store.py @@ -261,7 +261,7 @@ async def add_member( source_agent_id: str | None = None, memory_seed: str = "none", ) -> None: - if member_kind not in ("native", "clone"): + if member_kind not in ("native", "clone", "human"): raise ValueError(f"invalid member_kind: {member_kind}") if memory_seed not in ("none", "snapshot", "empty"): raise ValueError(f"invalid memory_seed: {memory_seed}") @@ -337,6 +337,64 @@ async def get_member(self, project_id: str, member_id: str) -> dict | None: keys = [d[0] for d in cur.description] return dict(zip(keys, row)) + async def is_project_member( + self, project_id: str, member_id: str, *, member_kind: str | None = None + ) -> bool: + """Return True if *member_id* is a member of *project_id*. + + When *member_kind* is provided, only match members of that kind + (e.g. ``"human"`` for cross-user collab delegation checks). + """ + if member_kind is not None: + async with self._db.execute( + "SELECT 1 FROM project_members " + "WHERE project_id = ? AND member_id = ? AND member_kind = ?", + (project_id, member_id, member_kind), + ) as cur: + return (await cur.fetchone()) is not None + else: + async with self._db.execute( + "SELECT 1 FROM project_members " + "WHERE project_id = ? AND member_id = ?", + (project_id, member_id), + ) as cur: + return (await cur.fetchone()) is not None + + async def get_project_setting( + self, project_id: str, key: str, default=None + ): + """Read a single key from the project's JSON settings dict. + + Returns *default* if the project does not exist, has no settings, or + the key is absent. + """ + project = await self.get_project(project_id) + if project is None: + return default + settings = project.get("settings") or {} + return settings.get(key, default) + + async def set_project_setting( + self, project_id: str, key: str, value + ) -> bool: + """Set a single key in the project's JSON settings dict. + + Returns True if the project was found and updated, False otherwise. + """ + project = await self.get_project(project_id) + if project is None: + return False + settings = project.get("settings") or {} + if not isinstance(settings, dict): + settings = {} + settings[key] = value + await self._db.execute( + "UPDATE projects SET settings = ?, updated_at = ? WHERE id = ?", + (json.dumps(settings), time.time(), project_id), + ) + await self._db.commit() + return True + 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 97bdd9f52..029b9b097 100644 --- a/tinyagentos/routes/agent_auth_requests.py +++ b/tinyagentos/routes/agent_auth_requests.py @@ -298,6 +298,7 @@ async def approve_request_record( decided_by: str, project_id: str | None = None, display_name: str | None = None, + sponsor_contact_id: str | None = None, ) -> dict: """Register an agent, mint its token, write grants + relationships + membership + a2a sync, and record the decision. @@ -310,6 +311,10 @@ async def approve_request_record( the consent route it is the approving admin's user_id, for invite auto-mode it is the invite's ``created_by``. + ``sponsor_contact_id`` is set for delegated agents (cross-user collab D1) + — the contact_id of the human who sponsored this agent. Omitted (None) for + normal consent and invite flows. + Returns ``{"status": "accepted", "canonical_id": ...}``. Raises ``HTTPException`` for the same guard failures as the consent route: @@ -397,6 +402,10 @@ async def approve_request_record( framework=record["framework"], project_id=project_id, ) + # For sponsored agents (cross-user collab D1), set the sponsor + # on the existing identity when reusing it for a new project. + if sponsor_contact_id: + await registry.set_sponsor(existing_cid, sponsor_contact_id) await add_agent_to_project( request, canonical_id=existing_cid, @@ -447,6 +456,7 @@ async def approve_request_record( # auth-request record / invite, not the registry origin column. origin="external-selfjoin", handle=handle, + sponsor_contact_id=sponsor_contact_id, ) canonical_id = reg_record["canonical_id"] diff --git a/tinyagentos/routes/decisions.py b/tinyagentos/routes/decisions.py index a2ed31bfe..1811dd7ab 100644 --- a/tinyagentos/routes/decisions.py +++ b/tinyagentos/routes/decisions.py @@ -317,7 +317,8 @@ async def answer_decision(decision_id: str, body: AnswerIn, request: Request, us routed_app = await _apply_app_grant(request, updated, body.value) routed_exec = await _apply_execution_grant(request, updated, body.value) routed_deleg = await _apply_delegation_grant(request, updated, body.value) - if not (routed_app or routed_exec or routed_deleg): + routed_collab = await _apply_collab_delegation_grant(request, updated, body.value) + if not (routed_app or routed_exec or routed_deleg or routed_collab): await _route_answer_to_agent(updated, body.value) return updated @@ -414,18 +415,63 @@ async def _apply_delegation_grant(request: Request, decision: dict, value) -> bo "delegate grant write failed for agent %s (decision %s)", from_agent, decision.get("id"), exc_info=True, ) - # Tell the agent the truth: only claim the task was assigned when the - # completion actually succeeded, so a failed assign is not reported as done. - if not approved: - reply = "delegation denied" - elif completed: - reply = "delegation approved - the task has been assigned" else: - reply = "delegation approved, but assigning the task failed - please retry" + reply = f"The delegation to {to_agent} was denied." await _route_answer_to_agent(decision, reply) return True +async def _apply_collab_delegation_grant(request: Request, decision: dict, value) -> bool: + """Side effect for a cross-user collab delegation-gate Decision (D1). + + The decision's metadata carries {kind: \"collab_delegation_gate\", contact_id, + agent_slug, display_name, granted_scopes, denied_scopes, project_id}. + + Approving it mints a project invite for the sponsored agent and returns + the connection bundle over the peer channel. Denying just routes the answer + (the remote contact's instance polls for the decision result). + + Mirrors ``_apply_delegation_grant``: the answer is already persisted, so a + delegation-completion hiccup must not fail the answer. + """ + meta = decision.get("metadata") or {} + if meta.get("kind") != "collab_delegation_gate": + return False + + approved = value == "approve" + if not approved: + # Denied -- just route the deny back so the remote contact knows. + deny_text = f"Delegation of {meta.get('agent_slug', '')} by {meta.get('contact_id', '')} was denied." + await _route_answer_to_agent(decision, deny_text) + return True + + # Approved -- mint the project invite via the delegation handler. + try: + from tinyagentos.delegation_handler import complete_delegation_approval + + result = await complete_delegation_approval( + request, decision_metadata=meta, + ) + if result.get("status") == "approved": + invite_id = result.get("invite_id", "") + reply = ( + f"Delegation approved. Agent '{meta.get('agent_slug', '')}' " + f"sponsored by {meta.get('contact_id', '')}. " + f"Invite ID: {invite_id}. " + f"Share this invite with the remote contact so their agent can join." + ) + await _route_answer_to_agent(decision, reply) + else: + error_text = f"Delegation approval failed: {result.get('error', 'unknown error')}" + await _route_answer_to_agent(decision, error_text) + except Exception: + logger.warning( + "collab delegation completion failed for decision %s", + decision.get("id"), exc_info=True, + ) + return True + + async def _apply_app_grant(request: Request, decision: dict, value) -> bool: """Side effect for an app-grant consent Decision: write the per-capability grant decisions to the app_grants ledger. The decision's metadata carries diff --git a/tinyagentos/routes/peer.py b/tinyagentos/routes/peer.py index 013dc73ff..cec20d16b 100644 --- a/tinyagentos/routes/peer.py +++ b/tinyagentos/routes/peer.py @@ -103,6 +103,7 @@ async def _authenticate_peer(request: Request) -> str: """Verify the bearer peer token and return the contact_id. Raises 401 if the token is missing, invalid, or the contact is not active. + Raises 503 if the per-instance peer panic switch is active. The peer token is stored hashed in peer_links.inbound_token_hash. We look up the contact via an indexed hash lookup on the inbound_token_hash @@ -112,6 +113,11 @@ async def _authenticate_peer(request: Request) -> str: if store is None: raise HTTPException(status_code=503, detail="peer channel not available") + # Per-instance panic kill-switch (D1 level 3): blocks all peer traffic. + from tinyagentos.delegation_handler import is_peer_disabled + if is_peer_disabled(request): + raise HTTPException(status_code=503, detail="peer routes disabled (per-instance panic)") + auth = request.headers.get("authorization", "") if not auth.lower().startswith("bearer "): raise HTTPException(status_code=401, detail="bearer token required") @@ -214,14 +220,23 @@ async def peer_inbox(body: PeerEnvelope, request: Request): # Mark peer as seen await store.mark_peer_seen(contact_id) - # Log the envelope kind for debugging; the real dispatch (collab invites, - # chat messages, etc.) happens in later milestones. + # Log the envelope kind for debugging; dispatch known kinds to their handlers. kind = envelope.get("kind", "unknown") logger.info( "peer_inbox: contact=%s kind=%s nonce=%s", contact_id, kind, envelope.get("nonce", "?"), ) + # Dispatch delegation requests to the cross-user collab handler (D1). + if kind == "delegation_request": + from tinyagentos.delegation_handler import process_delegation_request + + body_data = envelope.get("body", {}) + result = await process_delegation_request( + request, contact_id=contact_id, envelope_body=body_data, + ) + return {"status": result.get("status", "unknown"), "detail": result} + return {"status": "received", "kind": kind, "nonce": envelope.get("nonce")} diff --git a/tinyagentos/routes/project_invites.py b/tinyagentos/routes/project_invites.py index 45fa88f31..3c989378a 100644 --- a/tinyagentos/routes/project_invites.py +++ b/tinyagentos/routes/project_invites.py @@ -866,6 +866,15 @@ async def redeem_invite(request: Request, body: RedeemInviteIn): if "project_tasks" not in scopes: scopes = ["project_tasks"] + scopes + # Extract sponsor_contact_id from delegation-sponsored invites (D1). + invite_metadata = invite.get("metadata") or {} + if isinstance(invite_metadata, str): + try: + invite_metadata = json.loads(invite_metadata) + except (ValueError, TypeError): + invite_metadata = {} + sponsor_contact_id = invite_metadata.get("sponsor_contact_id") if isinstance(invite_metadata, dict) else None + handle = _derive_handle(project.get("slug") or invite["project_id"], body.harness, body.label) handle = await _dedupe_handle(request, handle) @@ -900,6 +909,7 @@ async def redeem_invite(request: Request, body: RedeemInviteIn): effective_project=invite["project_id"], decided_by=invite["created_by"], project_id=invite["project_id"], + sponsor_contact_id=sponsor_contact_id, ) except Exception as exc: # noqa: BLE001 - surface as JSON error await store.rollback_to_pending(body.invite_id) @@ -956,6 +966,15 @@ async def _redeem_os_level(request: Request, body: RedeemInviteIn, invite: dict, raw_scopes = [] scopes = list(raw_scopes) + # Extract sponsor_contact_id from delegation-sponsored invites (D1). + invite_metadata = invite.get("metadata") or {} + if isinstance(invite_metadata, str): + try: + invite_metadata = json.loads(invite_metadata) + except (ValueError, TypeError): + invite_metadata = {} + sponsor_contact_id = invite_metadata.get("sponsor_contact_id") if isinstance(invite_metadata, dict) else None + display_name = invite.get("display_name") handle = _derive_os_handle(display_name, body.harness, body.label) handle = await _dedupe_handle(request, handle) @@ -983,6 +1002,7 @@ async def _redeem_os_level(request: Request, body: RedeemInviteIn, invite: dict, decided_by=invite["created_by"], project_id=None, display_name=display_name, + sponsor_contact_id=sponsor_contact_id, ) except Exception as exc: # noqa: BLE001 - surface as JSON error await store.rollback_to_pending(body.invite_id) From cccdaf97eca92737e576a818991bc9a41c5ee4df Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Sun, 19 Jul 2026 15:01:03 +0200 Subject: [PATCH 2/3] fix(collab): resolve 5 Kilo findings on D1 delegation PR CRITICAL: Restore reply assignment in _apply_delegation_grant (regression from collapsed if/elif/else) WARNING: cascade_sponsor_revoke now filters agents by project_id via project_store.is_project_member for membership-scoped revoke WARNING: complete_delegation_approval re-applies scope denylist (validate_delegation_scopes) before minting invite SUGGESTION: Add kill_switch_reenable() for per-instance panic recovery SUGGESTION: A2A audit line emitted for contact-wide revoke across all projects the contact is a member of --- tinyagentos/delegation_handler.py | 89 ++++++++++++++++++++++++++----- tinyagentos/routes/decisions.py | 6 ++- 2 files changed, 81 insertions(+), 14 deletions(-) diff --git a/tinyagentos/delegation_handler.py b/tinyagentos/delegation_handler.py index 072777757..77afee108 100644 --- a/tinyagentos/delegation_handler.py +++ b/tinyagentos/delegation_handler.py @@ -325,10 +325,22 @@ async def complete_delegation_approval( if invite_store is None: return {"status": "error", "error": "invite store not available"} + # Re-apply scope denylist to ensure hard-denied scopes are stripped even + # if the stored metadata was tampered with between decision creation and + # approval. The denylist is the authoritative gate; the stored scopes are + # the human-approved set, but the code must never mint tokens for + # hard-denied scopes regardless. + safe_scopes, re_denied = validate_delegation_scopes(granted_scopes) + if re_denied: + logger.warning( + "complete_delegation_approval: re-stripped hard-denied scopes %r from stored grant", + re_denied, + ) + try: invite = await invite_store.mint( project_id=project_id, - scopes=granted_scopes, + scopes=safe_scopes, approval_mode="auto", check_interval_secs=1800, created_by=contact_id, @@ -363,9 +375,9 @@ async def cascade_sponsor_revoke( ) -> dict: """Revoke all sponsored agent identities for a contact. - When *project_id* is provided, only revoke tokens bound to that project - (membership-revoke). When None, revoke ALL sponsored identities - (contact-revoke). + When *project_id* is provided, only revoke agents that are members of + that project (membership-revoke). When None, revoke ALL sponsored + identities (contact-revoke). Also unassigns in-flight board tasks and posts an A2A system line. Returns a summary dict of what was revoked. @@ -374,12 +386,23 @@ async def cascade_sponsor_revoke( if registry is None: return {"status": "error", "error": "registry not available"} + project_store = getattr(request.app.state, "project_store", None) + # Find all active agents sponsored by this contact. sponsored = await registry.list_by_sponsor(contact_id, status="active") revoked_ids: list[str] = [] for agent in sponsored: canonical_id = agent["canonical_id"] + + # When scoped to a project, only revoke agents that are members of + # that project. Contact-wide revoke (project_id=None) hits all. + if project_id is not None and project_store is not None: + if not await project_store.is_project_member( + project_id, canonical_id + ): + continue + try: await registry.set_status(canonical_id, "revoked", actor=contact_id) revoked_ids.append(canonical_id) @@ -406,10 +429,19 @@ async def cascade_sponsor_revoke( canonical_id, exc_info=True, ) - # Post A2A system line. + # Post A2A system line. When project_id is set, post to that project's + # A2A channel. When None (contact-wide revoke), post to all projects the + # contact is a member of so every affected project sees the audit event. try: from tinyagentos.projects.a2a import ensure_a2a_channel + msg_store = request.app.state.chat_messages + msg = ( + f"Collaboration with {contact_id} has been revoked. " + f"{len(revoked_ids)} delegated agent(s) revoked, " + f"{unassigned_count} task(s) unassigned." + ) + if project_id: channel = await ensure_a2a_channel( request.app.state.chat_channels, @@ -417,17 +449,37 @@ async def cascade_sponsor_revoke( project_id, config=getattr(request.app.state, "config", None), ) - msg_store = request.app.state.chat_messages await msg_store.send_message( channel_id=channel["id"], author_id="system", author_type="user", - content=( - f"Collaboration with {contact_id} has been revoked. " - f"{len(revoked_ids)} delegated agent(s) revoked, " - f"{unassigned_count} task(s) unassigned." - ), + content=msg, ) + elif project_store is not None: + # Contact-wide: find all projects the contact is a member of. + try: + memberships = await project_store.list_projects_for_member(contact_id) + except AttributeError: + memberships = [] + for proj in (memberships or []): + pid = proj.get("id") or proj.get("project_id") + if not pid: + continue + try: + channel = await ensure_a2a_channel( + request.app.state.chat_channels, + request.app.state.project_store, + pid, + config=getattr(request.app.state, "config", None), + ) + await msg_store.send_message( + channel_id=channel["id"], + author_id="system", + author_type="user", + content=msg, + ) + except Exception: + pass except Exception: logger.warning( "cascade_revoke: A2A system line failed for %s", @@ -524,14 +576,25 @@ async def kill_switch_per_instance(request) -> dict: Sets a flag on app.state that the peer routes check on every request. Does NOT revoke tokens — just blocks all new peer traffic. - Reversible by clearing the flag. + Use ``kill_switch_reenable()`` to restore peer routes. """ - # Set a flag that the peer routes check. request.app.state._peer_disabled = True logger.warning("kill_switch: per-instance panic — peer routes DISABLED") return {"status": "panic", "peer_routes": "disabled"} +async def kill_switch_reenable(request) -> dict: + """Re-enable peer routes after a per-instance panic. + + Clears the panic flag set by ``kill_switch_per_instance``. + Does NOT re-establish revoked peer links or re-mint tokens — + those must be restored manually via the contacts/peer-link flow. + """ + request.app.state._peer_disabled = False + logger.info("kill_switch: per-instance panic cleared — peer routes RE-ENABLED") + return {"status": "recovered", "peer_routes": "enabled"} + + def is_peer_disabled(request) -> bool: """Check whether the per-instance peer panic switch is active.""" return getattr(request.app.state, "_peer_disabled", False) diff --git a/tinyagentos/routes/decisions.py b/tinyagentos/routes/decisions.py index 1811dd7ab..53c5aa4ae 100644 --- a/tinyagentos/routes/decisions.py +++ b/tinyagentos/routes/decisions.py @@ -415,8 +415,12 @@ async def _apply_delegation_grant(request: Request, decision: dict, value) -> bo "delegate grant write failed for agent %s (decision %s)", from_agent, decision.get("id"), exc_info=True, ) + if not approved: + reply = "delegation denied" + elif completed: + reply = "delegation approved - the task has been assigned" else: - reply = f"The delegation to {to_agent} was denied." + reply = "delegation approved, but assigning the task failed - please retry" await _route_answer_to_agent(decision, reply) return True From 68baed6ea5319d9d7046c91fde7dbb2ba05f1f44 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Sun, 19 Jul 2026 15:23:58 +0200 Subject: [PATCH 3/3] fix(collab): resolve 3 Kilo v2 findings on D1 delegation WARNING: Re-validate project membership at approval time in complete_delegation_approval (fail-closed runtime check per design spec 5) WARNING: Remove fragile list_projects_for_member call (method does not exist on project_store). Simplify A2A audit to project-scoped only. SUGGESTION: Route error answer back when complete_delegation_approval raises, so the remote contact receives feedback on approval failure. --- tinyagentos/delegation_handler.py | 77 +++++++++++++------------------ tinyagentos/routes/decisions.py | 5 ++ 2 files changed, 38 insertions(+), 44 deletions(-) diff --git a/tinyagentos/delegation_handler.py b/tinyagentos/delegation_handler.py index 77afee108..c77cd1c1a 100644 --- a/tinyagentos/delegation_handler.py +++ b/tinyagentos/delegation_handler.py @@ -321,6 +321,22 @@ async def complete_delegation_approval( if not all([contact_id, agent_slug, display_name, project_id]): return {"status": "error", "error": "incomplete decision metadata"} + # Re-validate project membership at approval time: the contact may have + # been removed from the project between decision creation and approval. + # This is the fail-closed runtime check required by section 5 of the + # cross-user-collab design spec. + project_store = getattr(request.app.state, "project_store", None) + if project_store is not None and not await project_store.is_project_member( + project_id, contact_id, member_kind="human" + ): + return { + "status": "error", + "error": ( + f"contact {contact_id} is no longer a human collaborator " + f"on project {project_id}" + ), + } + invite_store = getattr(request.app.state, "project_invite_store", None) if invite_store is None: return {"status": "error", "error": "invite store not available"} @@ -429,20 +445,14 @@ async def cascade_sponsor_revoke( canonical_id, exc_info=True, ) - # Post A2A system line. When project_id is set, post to that project's - # A2A channel. When None (contact-wide revoke), post to all projects the - # contact is a member of so every affected project sees the audit event. - try: - from tinyagentos.projects.a2a import ensure_a2a_channel - - msg_store = request.app.state.chat_messages - msg = ( - f"Collaboration with {contact_id} has been revoked. " - f"{len(revoked_ids)} delegated agent(s) revoked, " - f"{unassigned_count} task(s) unassigned." - ) + # Post A2A system line to the affected project (when scoped). + # Contact-wide revoke does not emit A2A — each project's membership-revoke + # path handles its own audit event when the cascade fires per-project. + if project_id: + try: + from tinyagentos.projects.a2a import ensure_a2a_channel - if project_id: + msg_store = request.app.state.chat_messages channel = await ensure_a2a_channel( request.app.state.chat_channels, request.app.state.project_store, @@ -453,38 +463,17 @@ async def cascade_sponsor_revoke( channel_id=channel["id"], author_id="system", author_type="user", - content=msg, + content=( + f"Collaboration with {contact_id} has been revoked. " + f"{len(revoked_ids)} delegated agent(s) revoked, " + f"{unassigned_count} task(s) unassigned." + ), + ) + except Exception: + logger.warning( + "cascade_revoke: A2A system line failed for %s", + contact_id, exc_info=True, ) - elif project_store is not None: - # Contact-wide: find all projects the contact is a member of. - try: - memberships = await project_store.list_projects_for_member(contact_id) - except AttributeError: - memberships = [] - for proj in (memberships or []): - pid = proj.get("id") or proj.get("project_id") - if not pid: - continue - try: - channel = await ensure_a2a_channel( - request.app.state.chat_channels, - request.app.state.project_store, - pid, - config=getattr(request.app.state, "config", None), - ) - await msg_store.send_message( - channel_id=channel["id"], - author_id="system", - author_type="user", - content=msg, - ) - except Exception: - pass - except Exception: - logger.warning( - "cascade_revoke: A2A system line failed for %s", - contact_id, exc_info=True, - ) return { "status": "revoked", diff --git a/tinyagentos/routes/decisions.py b/tinyagentos/routes/decisions.py index 53c5aa4ae..2832093ab 100644 --- a/tinyagentos/routes/decisions.py +++ b/tinyagentos/routes/decisions.py @@ -473,6 +473,11 @@ async def _apply_collab_delegation_grant(request: Request, decision: dict, value "collab delegation completion failed for decision %s", decision.get("id"), exc_info=True, ) + error_msg = ( + f"Delegation approval failed: an internal error occurred while " + f"setting up the sponsored agent. Please retry or check the logs." + ) + await _route_answer_to_agent(decision, error_msg) return True