From 6c507c17206cfee7eb7853d443a939f37bf49854 Mon Sep 17 00:00:00 2001 From: jaylfc Date: Tue, 28 Jul 2026 18:44:33 +0000 Subject: [PATCH] tsk-2wbp3f [OPEN] Delegation A3: deliver the minted invite to the re --- tests/test_collab_d1_delegation.py | 300 ++++++++++++ tinyagentos/agent_registry_store.py | 104 ++++- tinyagentos/delegation_handler.py | 537 ++++++++++++++++++++++ tinyagentos/projects/invite_store.py | 61 ++- tinyagentos/projects/project_store.py | 58 +++ tinyagentos/routes/agent_auth_requests.py | 10 + tinyagentos/routes/decisions.py | 62 ++- tinyagentos/routes/peer.py | 20 +- tinyagentos/routes/project_invites.py | 20 + 9 files changed, 1124 insertions(+), 48 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 017c9a63d..6438c66ee 100644 --- a/tinyagentos/projects/invite_store.py +++ b/tinyagentos/projects/invite_store.py @@ -32,7 +32,8 @@ display_name TEXT, kind TEXT NOT NULL DEFAULT 'agent', pin_required INTEGER NOT NULL DEFAULT 1, - contact_id TEXT + contact_id TEXT, + metadata TEXT NOT NULL DEFAULT '{}' ); """ @@ -110,6 +111,11 @@ async def _post_init(self) -> None: "ALTER TABLE project_invites ADD COLUMN contact_id 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)" @@ -132,6 +138,7 @@ async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, kind: str = "agent", pin_required: bool = True, contact_id: str | None = None, + metadata: dict | None = None, ttl_secs: int | None = None) -> dict: if self._db is None: raise RuntimeError("ProjectInviteStore not initialised") @@ -195,26 +202,27 @@ async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, await self._db.execute( """ 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, kind, pin_required, contact_id) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?) - """, - ( - invite_id, - project_id, - pin_hash, - json.dumps(scopes), - approval_mode, - check_interval_secs, - created_by, - now, - expires_ts, - display_name, - kind, - int(pin_required), - contact_id, - ), + (invite_id, project_id, pin_hash, scopes, approval_mode, + check_interval_secs, created_by, created_ts, expires_ts, status, + display_name, kind, pin_required, contact_id, metadata) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?) + """, + ( + invite_id, + project_id, + pin_hash, + json.dumps(scopes), + approval_mode, + check_interval_secs, + created_by, + now, + expires_ts, + display_name, + kind, + int(pin_required), + contact_id, + json.dumps(metadata or {}), + ), ) await self._db.commit() break @@ -249,6 +257,7 @@ async def mint(self, *, project_id=None, scopes: list[str], approval_mode: str, "kind": kind, "pin_required": int(pin_required), "contact_id": contact_id, + "metadata": metadata or {}, }, "pin": pin, } @@ -458,4 +467,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 f796f4a1d..06d057bcc 100644 --- a/tinyagentos/projects/project_store.py +++ b/tinyagentos/projects/project_store.py @@ -379,6 +379,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 5020ca343..5d2609b77 100644 --- a/tinyagentos/routes/agent_auth_requests.py +++ b/tinyagentos/routes/agent_auth_requests.py @@ -355,6 +355,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. @@ -367,6 +368,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: @@ -452,6 +457,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, @@ -502,6 +511,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 a62342335..cb15d3728 100644 --- a/tinyagentos/routes/peer.py +++ b/tinyagentos/routes/peer.py @@ -112,6 +112,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 @@ -121,6 +122,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") @@ -240,7 +246,7 @@ async def peer_inbox(body: PeerEnvelope, request: Request): # Mark peer as seen await store.mark_peer_seen(contact_id) - # Dispatch the envelope by kind. + # Log the envelope kind for debugging; dispatch known kinds to their handlers. kind = envelope.get("kind", "unknown") body_data = envelope.get("body", {}) @@ -250,12 +256,20 @@ async def peer_inbox(body: PeerEnvelope, request: Request): if kind in ("collab_invite_accept", "collab_invite_decline"): return await _handle_collab_response(request, contact_id, envelope, body_data, kind) + # Dispatch delegation requests to the cross-user collab handler (D1). + if kind == "delegation_request": + from tinyagentos.delegation_handler import process_delegation_request + + result = await process_delegation_request( + request, contact_id=contact_id, envelope_body=body_data, + ) + return {"status": result.get("status", "unknown"), "detail": result} + # Log unrecognised kinds for debugging; they are accepted but not dispatched. logger.info( - "peer_inbox: contact=%s kind=%s nonce=%s (unrecognised kind — accepted, no dispatch)", + "peer_inbox: contact=%s kind=%s nonce=%s (unrecognised kind -- accepted, no dispatch)", contact_id, kind, envelope.get("nonce", "?"), ) - return {"status": "received", "kind": kind, "nonce": envelope.get("nonce")} diff --git a/tinyagentos/routes/project_invites.py b/tinyagentos/routes/project_invites.py index 58da40cc7..cef44a543 100644 --- a/tinyagentos/routes/project_invites.py +++ b/tinyagentos/routes/project_invites.py @@ -1121,6 +1121,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) @@ -1155,6 +1164,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) @@ -1211,6 +1221,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) @@ -1238,6 +1257,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)