From d1b43613d6ffdc29ef9ddac3b7fe10380cc58d02 Mon Sep 17 00:00:00 2001 From: Jay Moran Date: Tue, 8 Sep 2026 15:47:28 -0700 Subject: [PATCH 01/57] Stop requiring regeneration for cells with no declared design The notebook/SDK flow PUTs cells straight through upsert_replicate with no design_spec, so there are no planned labels to compare against and no design to regenerate -- but the label comparison reported drift anyway, permanently 422'ing "run all cells" on every notebook-driven experiment. Gate that comparison on a declared factorial matrix; a design that had factors and no longer declares them still trips the spec comparison. Also fixes _cells_of's MissingGreenlet: a replicate's interesting attributes read through .cell, which a lazy load can't populate under asyncio. --- src/asaree/services/design_generation.py | 13 ++++++- tests/test_design_revisions.py | 44 ++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/src/asaree/services/design_generation.py b/src/asaree/services/design_generation.py index 663f36a..3c18553 100644 --- a/src/asaree/services/design_generation.py +++ b/src/asaree/services/design_generation.py @@ -206,9 +206,20 @@ async def get_design_impact( current_cell_keys = { _cell_key(replicate.factor_values, replicate.replicate_label) for replicate in current_replicates } + # The label comparison only means something when a factorial matrix is + # actually declared. An experiment that never declared one but has cells is + # the supported notebook/SDK flow -- cells PUT straight through + # ``upsert_replicate`` onto the revision ``get_or_create_current`` opens for + # them (see services/design_revisions.py's module docstring). There is + # nothing planned to compare those labels against, so requiring a + # regeneration would permanently block "run all cells" on an experiment + # that has no design to regenerate. Genuine drift -- including a design + # whose final factor was removed without regenerating -- still trips the + # spec comparison on the left. + drifted_labels = bool(material["factors"]) and current_labels != planned_labels return DesignImpact( has_generated_design=True, - regeneration_required=material_design_spec(current.design_spec) != material or current_labels != planned_labels, + regeneration_required=material_design_spec(current.design_spec) != material or drifted_labels, current_cell_count=len(current_cell_keys), proposed_cell_count=len(planned_cell_keys), added_cell_count=len(planned_cell_keys - current_cell_keys), diff --git a/tests/test_design_revisions.py b/tests/test_design_revisions.py index a5e65fc..21f85d8 100644 --- a/tests/test_design_revisions.py +++ b/tests/test_design_revisions.py @@ -191,6 +191,43 @@ async def test_design_impact_previews_an_expansion_before_regeneration(experimen assert (impact.added_replicate_count, impact.retained_replicate_count, impact.removed_replicate_count) == (4, 2, 0) +async def test_cells_without_a_declared_design_never_require_regeneration(experiment_id: uuid.UUID) -> None: + """The notebook/SDK flow: cells PUT straight through upsert_replicate onto + the revision get_or_create_current opens for them, with no design_spec ever + declared (see services/design_revisions.py's module docstring). + + Nothing is planned, so there are no labels to compare against and no design + to regenerate. Reporting regeneration_required here permanently 422'd + "run all cells" on every notebook-driven experiment.""" + async with get_session() as db: + for label in ("cell-1", "cell-2"): + await upsert_replicate(db, experiment_id=experiment_id, replicate_label=label, fields={"factor_values": {}}) + + async with get_session() as db: + impact = await get_design_impact(db, experiment_id=experiment_id, design_spec=None) + + assert impact.regeneration_required is False + assert (impact.current_replicate_count, impact.proposed_replicate_count) == (2, 0) + + +async def test_removing_the_final_factor_without_regenerating_still_requires_regeneration( + experiment_id: uuid.UUID, +) -> None: + """The case the empty-factors allowance must not swallow: a design that + *had* factors and no longer declares them has genuinely drifted from its + materialized cells, and the revision's own recorded spec is what proves it.""" + async with get_session() as db: + await generate_design_cells( + db, experiment_id=experiment_id, factors=_TWO_BY_ONE, design_spec={"factors": _TWO_BY_ONE} + ) + + async with get_session() as db: + impact = await get_design_impact(db, experiment_id=experiment_id, design_spec={"factors": []}) + + assert impact.regeneration_required is True + assert (impact.current_replicate_count, impact.proposed_replicate_count) == (2, 0) + + async def test_design_counts_cells_separately_from_replicates(experiment_id: uuid.UUID) -> None: async with get_session() as db: await generate_design_cells( @@ -313,9 +350,16 @@ async def test_revision_numbers_are_never_reused(experiment_id: uuid.UUID) -> No def _cells_of(experiment_id: uuid.UUID): # type: ignore[no-untyped-def] from sqlalchemy import select + from sqlalchemy.orm import contains_eager + # contains_eager, not a bare join: every interesting attribute of a + # replicate (design_revision_id, experiment_id, factor_values) is a property + # that reads through .cell, and a lazy load can't fire from sync property + # access under asyncio -- it raises MissingGreenlet. The join is already + # there; this just tells the ORM to populate the relationship from it. return ( select(FactorialReplicateResult) .join(FactorialReplicateResult.cell) + .options(contains_eager(FactorialReplicateResult.cell)) .where(FactorialCell.experiment_id == experiment_id) ) From 12799b1c991168c89564ce7549154644408577d6 Mon Sep 17 00:00:00 2001 From: Jay Moran Date: Tue, 8 Sep 2026 15:49:28 -0700 Subject: [PATCH 02/57] Derive peer agent cards and mediate consultations Two Agent nodes joined by a plain canvas edge can now see each other and choose to work together, with no MCP server and no user-wired Tool node. This is the ASAREE half of the capability core gained in Motoro: core projects and dispatches, ASAREE decides who exists and authorizes every call. An AgentCard is derived, never stored. Capability comes from the published protocol revision the run is pinned to (so a canvas edit cannot change what a queued run believes about its peers) while reachability is read live from the draft graph on every call -- an unplugged edge stops a consultation mid-run. resolve_available_agents walks the graph for peers; _is_peer_edge is the one definition of what a peer edge is (a plain non-connector edge between two Agent nodes, read undirected, no self-peer). AgentMessenger is where the invariants live, because the model must not be able to state any of them. It authorizes the recipient against the live graph, assigns sender identity, message ids and sequence, appends the transcript, runs the peer's turn to completion, and returns its answer as the function call's result. There is deliberately no send_message/finish action protocol: an agent finishes by writing its final answer exactly as a single agent does today. The only new thing in its world is that a peer exists and can be asked, which is what keeps single-agent runs byte-for-byte unchanged -- no peers means no field content, no prompt section, no function schema. Budgets are caps on executions, nesting depth and wall clock, and exhausting one returns a *call result*, not an exception, so the caller can still finish with a real answer -- a run that hit a limit is finished, not failed. The wall clock pauses rather than extends (services/deadline.py): a consulting agent is blocked on its peer, so charging it for time it could not use would make asking for help a reason to time out. A ContextVar chain of Deadlines means the peer's own limit still applies to the peer. The transcript persists on ProtocolRun.conversation (migration ba2c3d4e5f60) in A2A's Message/Part shape, so a future data or file part is a new branch rather than a migration. Also states the rule on _sync_durable_agent that this design depends on: nothing run-scoped may enter its fields. update_agent reads None as "leave unchanged", so a field written once during one run is never cleared by any later one. Covered by tests/test_agent_cards.py and tests/test_agent_messenger.py. --- ...c3d4e5f60_add_protocol_run_conversation.py | 27 ++ src/asaree/models/protocol_run.py | 11 + src/asaree/services/agent_cards.py | 132 ++++++ src/asaree/services/agent_messenger.py | 445 ++++++++++++++++++ src/asaree/services/deadline.py | 118 +++++ src/asaree/services/protocol_execution.py | 252 +++++++++- src/asaree/services/protocol_runs.py | 27 +- tests/test_agent_cards.py | 236 ++++++++++ tests/test_agent_messenger.py | 404 ++++++++++++++++ 9 files changed, 1641 insertions(+), 11 deletions(-) create mode 100644 src/asaree/migrations/versions/ba2c3d4e5f60_add_protocol_run_conversation.py create mode 100644 src/asaree/services/agent_cards.py create mode 100644 src/asaree/services/agent_messenger.py create mode 100644 src/asaree/services/deadline.py create mode 100644 tests/test_agent_cards.py create mode 100644 tests/test_agent_messenger.py diff --git a/src/asaree/migrations/versions/ba2c3d4e5f60_add_protocol_run_conversation.py b/src/asaree/migrations/versions/ba2c3d4e5f60_add_protocol_run_conversation.py new file mode 100644 index 0000000..1a7f42a --- /dev/null +++ b/src/asaree/migrations/versions/ba2c3d4e5f60_add_protocol_run_conversation.py @@ -0,0 +1,27 @@ +"""add persisted agent-to-agent conversation transcript + +Revision ID: ba2c3d4e5f60 +Revises: b8c9d0e1f2a3 +Create Date: 2026-09-08 00:00:00.000000 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "ba2c3d4e5f60" +down_revision: str | None = "b8c9d0e1f2a3" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column("protocol_runs", sa.Column("conversation", postgresql.JSONB(), nullable=True), if_not_exists=True) + + +def downgrade() -> None: + op.drop_column("protocol_runs", "conversation", if_exists=True) diff --git a/src/asaree/models/protocol_run.py b/src/asaree/models/protocol_run.py index 7f99ef4..56e6317 100644 --- a/src/asaree/models/protocol_run.py +++ b/src/asaree/models/protocol_run.py @@ -48,6 +48,17 @@ class ProtocolRun(Base, TimestampMixin): # erasing the scores/evaluation state a user may inspect on an older run. # Shape: {"metric_values": {...}, "metric_evaluation": {...}}. attempt_result: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) + # The agent-to-agent transcript, present only once a run's agents actually + # consult each other -- null for every single-agent and pipeline run. + # Shape: {"state": , "messages": [{"message_id", "sequence", + # "from_agent_id", "to_agent_id", "parts": [...], "created_at"}, ...]} + # Kept alongside node_runs and for the same reason: it is an append-only, + # run-scoped document that only the polling endpoint and the canvas read, + # and both read it whole. A messages table would add joins without + # supporting a query the product needs. `sequence` is stored rather than + # inferred from list position so an entry keeps its identity if the + # document is ever paged or filtered. + conversation: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) # Protocol-level failure (e.g. a cycle rejected at validation time, or an # unhandled executor exception) -- distinct from any one node's own error # already recorded inside node_runs. diff --git a/src/asaree/services/agent_cards.py b/src/asaree/services/agent_cards.py new file mode 100644 index 0000000..23dac18 --- /dev/null +++ b/src/asaree/services/agent_cards.py @@ -0,0 +1,132 @@ +"""The AgentCard: what one Agent node looks like to another. + +An AgentCard answers "who is this agent and what is it for", in a form another +agent can read and act on. A2A defines it as the capability-discovery document; +here it plays the same role, minus the network -- see +``local_files/agent-communication/01-agent-card.md`` and ``06-a2a-alignment.md``. + +**It is derived, never stored.** There is no card table, no card column and no +card cache. A card is a pure projection of an Agent node plus its connectors, +computed at run start by the same pass that already computes ``model_config`` / +``tool_config`` / ``skill_config`` / ``pattern_config`` from the graph. That is +precisely what makes it dynamic: wiring a Skill node takes effect on the next run +today with nothing to invalidate, and the card inherits that for free. If you +find yourself adding a place to persist one, the staleness problem comes back. + +This module owns the card's *shape* only. The graph reading that feeds it lives +in :mod:`asaree.services.protocol_execution`, next to every other ``_resolve_*`` +-- which is also what keeps the import one-directional. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass(frozen=True) +class AgentSkill: + """One thing this agent can do. A2A's ``AgentCard.skills`` entry. + + ``id`` is a slug of the name rather than the registered skill's UUID: it + only has to be unique within one card, and the reader is a language model + for which a UUID is noise. Nothing dereferences it -- a peer cannot load + another agent's skill. + """ + + id: str + name: str + description: str + + def to_dict(self) -> dict[str, str]: + return {"id": self.id, "name": self.name, "description": self.description} + + +@dataclass(frozen=True) +class AgentCard: + """A2A-shaped capability descriptor for one Agent node. + + ``agent_id`` is the **canvas node id** -- never the Motoro ``Agent.id``, and + never the label. The node id is what graph connectivity authorizes against, + what the transcript stores, and what stays stable across re-syncs of the + durable Motoro agent; the Motoro agent's UUID is an implementation detail of + one execution and must not leak into the card. + + Deliberately absent: tools and MCP servers (a peer that knows which servers + another agent can reach will ask it to proxy a call -- out of scope, and a + permission hole), the system prompt (user-authored, long, and mostly + meaningless to another agent), and endpoint/auth/security schemes (A2A + carries those because its peers are remote and untrusted; in-process peers + in one graph owned by one user have no endpoint to publish). + """ + + agent_id: str + name: str + description: str + skills: list[AgentSkill] = field(default_factory=list) + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + """The wire shape Motoro's ``available_agents`` consumes. + + Plain JSON-able dicts rather than the dataclass itself, because this + crosses into core -- which must not learn an ASAREE type -- and is + snapshotted into ``RunContext`` for resume. + """ + return { + "agent_id": self.agent_id, + "name": self.name, + "description": self.description, + "skills": [s.to_dict() for s in self.skills], + "metadata": dict(self.metadata), + } + + +def _skill_slug(name: str) -> str: + return "".join(c if c.isalnum() else "-" for c in name.strip().lower()).strip("-") or "skill" + + +def build_agent_card( + *, + node_id: str, + label: str | None, + description: str, + goal: str, + skills: list[dict[str, Any]], + model: str | None, + metadata: dict[str, Any] | None = None, +) -> AgentCard: + """Assemble a card from values the caller has already resolved. + + Pure on purpose: every input here is something ``_run_agent_node`` reads out + of the graph anyway, so the card cannot disagree with the agent that + actually runs. + + *description* falls back to *goal* and then to a generated default, because + the description is the single highest-leverage field for making a peer's + model choose well -- it becomes the consultation function's description in + ``motoro.engine.agent_channel``. An empty one is worse than a generic one. + """ + name = (label or "").strip() or node_id + text = (description or "").strip() or (goal or "").strip() or f"An agent named {name}." + card_metadata: dict[str, Any] = dict(metadata or {}) + if model: + card_metadata["model"] = model + return AgentCard( + agent_id=node_id, + name=name, + description=text, + skills=[ + AgentSkill( + id=_skill_slug(str(s.get("name") or "")), + name=str(s.get("name") or ""), + description=str(s.get("description") or ""), + ) + for s in skills + if s.get("name") + ], + metadata=card_metadata, + ) + + +__all__ = ["AgentCard", "AgentSkill", "build_agent_card"] diff --git a/src/asaree/services/agent_messenger.py b/src/asaree/services/agent_messenger.py new file mode 100644 index 0000000..8012086 --- /dev/null +++ b/src/asaree/services/agent_messenger.py @@ -0,0 +1,445 @@ +"""Delivery of one agent's question to a connected peer, and the run that hosts it. + +Motoro's :class:`~motoro.engine.ports.AgentMessengerPort` is deliberately one +method wide: the engine projects the peers a caller declared into callable +function schemas and hands any resulting call straight back here. Everything +that decides *whether* the call happens lives in this module -- authorization, +message identity and ordering, the transcript, the budget, and cancellation. +The engine never learns what an ASAREE canvas is. + +**The reply is a call result, not an exception.** A peer that may not be +reached, a spent budget and a cancelled conversation all come back as an +:class:`~motoro.engine.ports.AgentReply` with a state the calling model can +read, so it absorbs the outcome and still writes a real answer. Only genuine +infrastructure failure raises. + +**Capability is snapshotted; reachability is live.** A peer's card describes it +as the published revision configures it, so a canvas edit cannot hot-patch a +run's agents mid-flight. Authorization is re-asked of the *draft* graph on every +single consultation, so pulling the edge on the canvas stops the next +consultation immediately. Two different questions, deliberately reading two +different graphs. + +This module imports the executor, never the reverse. A pipeline run knows +nothing about conversations, which is what keeps invariant 11 (single-agent runs +are byte-for-byte unaffected) structural rather than a promise. +""" + +from __future__ import annotations + +import logging +import time +import uuid +from datetime import UTC, datetime, timedelta +from typing import Any + +from motoro.engine.ports import AgentReply + +from asaree.models.database import get_session +from asaree.services.deadline import deadlines_paused +from asaree.services.experiments import get_experiment +from asaree.services.protocol_execution import ( + _AGENT_CANCELLED, + _ambient_meta_for, + _can_deliver_communication, + _compute_workspace_id, + _run_agent_node, + resolve_available_agents, +) +from asaree.services.protocol_revisions import get_revision +from asaree.services.protocol_runs import get_protocol_run, set_status, update_conversation, update_node_run +from asaree.services.protocols import get_protocol + +logger = logging.getLogger(__name__) + +#: Total peer turns allowed per protocol run, across the whole conversation +#: tree. Every one of these is a full agent run with its own Reason/Plan/Act +#: cycle and its own tokens, so this is a cost cap as much as a loop guard. +_MAX_PEER_EXECUTIONS = 8 + +#: The real backstop. Deliberately **not** extended by peer time the way an +#: individual agent's own deadline is (see :mod:`asaree.services.deadline`): +#: no agent is charged for delegating, but the total stays bounded. +_MAX_CONVERSATION_DURATION = timedelta(minutes=5) + +#: How deep consultations may nest. Two is enough for "Planner asks Critic, +#: Critic asks a clarifying question back" -- the shape this feature exists for +#: -- without letting a chain of agents each delegate one level further. +_MAX_CONSULT_DEPTH = 2 + +#: The transcript's stand-in sender for the human who started the conversation. +#: Not a node id, and deliberately not a valid one: nothing can address it, and +#: authorization would refuse if anything tried. +USER_PARTICIPANT = "user" + + +def _text_of(parts: list[dict[str, Any]]) -> str: + return "\n".join(str(p.get("text", "")) for p in parts if p.get("kind") == "text").strip() + + +class AgentMessenger: + """One per protocol run. Owns that run's whole agent-to-agent surface. + + Holds the conversation document in memory as the authoritative copy and + checkpoints it to ``ProtocolRun.conversation`` around every peer execution, + which is what lets a worker retry see exactly how far it got. + + Not concurrency-safe, by design: invariant 5 is that one agent executes at a + time and a consulting agent blocks on its peer's reply, so ``sequence`` and + the budget counters are only ever touched from a single logical call stack. + """ + + def __init__( + self, + *, + protocol_id: uuid.UUID, + protocol_run_id: uuid.UUID, + owner_id: uuid.UUID, + graph: dict[str, Any], + entry_agent_id: str, + workspace_id: str | None = None, + ) -> None: + #: The *revision* graph -- capability: who each peer is and how it runs. + #: Authorization reads the live draft graph instead, on every call. + self._graph = graph + self._protocol_id = protocol_id + self._protocol_run_id = protocol_run_id + self._owner_id = owner_id + self._workspace_id = workspace_id + self._entry_agent_id = entry_agent_id + self._started_at = time.monotonic() + self._executions = 0 + self._depth = 0 + self._sequence = 0 + self._messages: list[dict[str, Any]] = [] + self._state = "working" + #: Set when a cap is what stopped the conversation, so the run can land + #: on ``limit_reached`` rather than looking like a clean completion. + self.limit_reached = False + + # -- transcript ---------------------------------------------------- + + @property + def conversation(self) -> dict[str, Any]: + """The transcript as stored: one JSON document, read whole.""" + return {"state": self._state, "entry_agent_id": self._entry_agent_id, "messages": list(self._messages)} + + def set_state(self, state: str) -> None: + self._state = state + + def append( + self, + *, + from_agent_id: str, + to_agent_id: str, + parts: list[dict[str, Any]], + state: str | None = None, + ) -> dict[str, Any]: + """Record one message with a runtime-assigned id and ``sequence``. + + Invariant 2: identity and ordering are assigned here, never taken from + model output. ``state`` is set on replies only -- a request has no + outcome of its own yet. + """ + self._sequence += 1 + message = { + "message_id": str(uuid.uuid4()), + "sequence": self._sequence, + "from_agent_id": from_agent_id, + "to_agent_id": to_agent_id, + "parts": parts, + "created_at": datetime.now(UTC).isoformat(), + **({"state": state} if state is not None else {}), + } + self._messages.append(message) + return message + + async def checkpoint(self) -> None: + async with get_session() as db: + await update_conversation(db, self._protocol_run_id, self.conversation) + + # -- delivery ------------------------------------------------------ + + async def send( + self, + *, + from_agent_id: str, + to_agent_id: str, + parts: list[dict[str, Any]], + context: Any = None, + ) -> AgentReply: + """Deliver one question to a peer and return its reply. + + The request is recorded *before* any refusal check runs, so a rejected + consultation still appears in the transcript (invariant 10) -- a + silently dropped question is exactly the failure this makes debuggable. + + ``context`` is the engine's ``RunContext``. Unused: everything this + needs is per-protocol-run state held on the instance, and reading run + state out of the engine's context would be a second source of truth. + """ + self.append(from_agent_id=from_agent_id, to_agent_id=to_agent_id, parts=parts) + + refusal = await self._refusal(from_agent_id, to_agent_id) + if refusal is not None: + logger.info("consultation refused (%s -> %s): %s", from_agent_id, to_agent_id, refusal) + return await self._reply(to_agent_id, from_agent_id, refusal, state="rejected") + + await self.checkpoint() + + self._executions += 1 + self._depth += 1 + try: + # The caller's own clock stops for exactly this span, failures + # included: it waited either way, and charging it for a peer's + # failure is the same unfairness as charging it for a peer's + # success. The conversation-level cap above keeps ticking. + with deadlines_paused(): + output_text, error, run_id = await self._run_peer(to_agent_id, parts) + finally: + self._depth -= 1 + + if error == _AGENT_CANCELLED: + return await self._reply( + to_agent_id, from_agent_id, "The consultation was cancelled.", state="canceled", task_id=run_id + ) + if error is not None: + return await self._reply( + to_agent_id, + from_agent_id, + f"The agent could not answer: {error}", + state="failed", + task_id=run_id, + error=error, + ) + text = (output_text or "").strip() + return await self._reply( + to_agent_id, + from_agent_id, + text or "The agent finished without producing an answer.", + state="completed", + task_id=run_id, + ) + + async def _reply( + self, + from_agent_id: str, + to_agent_id: str, + text: str, + *, + state: str, + task_id: str | None = None, + error: str | None = None, + ) -> AgentReply: + parts = [{"kind": "text", "text": text}] + self.append(from_agent_id=from_agent_id, to_agent_id=to_agent_id, parts=parts, state=state) + await self.checkpoint() + return AgentReply(state=state, parts=parts, task_id=task_id, error=error) + + async def _refusal(self, from_agent_id: str, to_agent_id: str) -> str | None: + """The reason this consultation may not proceed, or ``None``. + + Every branch returns prose addressed to the calling *model*, because + that is who reads it: it has to be able to tell "you may not ask this + agent" from "you have asked enough" and choose differently, so a bare + "rejected" would be worse than useless. + """ + if self._depth >= _MAX_CONSULT_DEPTH: + self.limit_reached = True + return ( + f"Consultations are already nested {_MAX_CONSULT_DEPTH} deep, which is the limit. " + "Answer with what you have rather than delegating further." + ) + if self._executions >= _MAX_PEER_EXECUTIONS: + self.limit_reached = True + return ( + f"This conversation has used all {_MAX_PEER_EXECUTIONS} of its peer consultations. " + "Answer with what you already have." + ) + if time.monotonic() - self._started_at >= _MAX_CONVERSATION_DURATION.total_seconds(): + self.limit_reached = True + return ( + f"This conversation has run for its full {int(_MAX_CONVERSATION_DURATION.total_seconds())} seconds. " + "Answer with what you already have." + ) + + async with get_session() as db: + run = await get_protocol_run(db, self._protocol_run_id) + if run is not None and run.cancel_requested_at is not None: + # Invariant 8: a Stop seen between consultations stops further + # ones. An already in-flight peer run is separately interrupted + # by _execute_run_cancellable's own poller. + return "This run was cancelled, so the consultation was not delivered." + protocol = await get_protocol(db, self._protocol_id) + + # The live draft graph, not the revision this run's agents came from. + live_graph = protocol.graph if protocol is not None else self._graph + if not _can_deliver_communication(live_graph, from_agent_id, to_agent_id): + return ( + "That agent is not connected to you on the canvas, so it cannot be consulted. " + "Answer using the agents listed for you, or with what you already have." + ) + return None + + async def _run_peer( + self, to_agent_id: str, parts: list[dict[str, Any]] + ) -> tuple[str | None, str | None, str | None]: + """Give the peer its own full turn. + + A real nested agent run, not a prompt trick: its own Motoro ``AgentRun``, + so cost, steps and the Runs tab attribute it separately, and its own + ``available_agents`` so it may consult back within the depth cap. + """ + node = next((n for n in self._graph.get("nodes") or [] if str(n.get("id")) == to_agent_id), None) + if node is None: + # Only reachable if the live graph and the revision disagree about + # whether a node exists, which authorization above cannot catch. + return None, "the agent no longer exists in this protocol revision", None + + async with get_session() as db: + await update_node_run(db, self._protocol_run_id, to_agent_id, {"status": "running"}) + + output_text, error, run_id = await _run_agent_node( + node, + protocol_id=self._protocol_id, + protocol_run_id=self._protocol_run_id, + owner_id=self._owner_id, + user_input=_text_of(parts), + graph=self._graph, + workspace_id=self._workspace_id, + ambient_meta=_ambient_meta_for(self._graph, to_agent_id, self._workspace_id), + available_agents=await resolve_available_agents(self._graph, to_agent_id, owner_id=self._owner_id), + agent_messenger=self, + ) + # The canvas shows a consulted peer as a node that ran, because it did. + # A peer consulted twice keeps only its latest turn here; the full + # sequence is the transcript's job, not node_runs'. + async with get_session() as db: + await update_node_run( + db, + self._protocol_run_id, + to_agent_id, + { + "status": "cancelled" if error == _AGENT_CANCELLED else ("failed" if error else "completed"), + "output_text": output_text, + "error": None if error == _AGENT_CANCELLED else error, + "run_id": str(run_id) if run_id else None, + }, + ) + return output_text, error, str(run_id) if run_id else None + + +async def run_conversation(protocol_run_id: uuid.UUID, *, entry_agent_id: str, user_input: str) -> None: + """Execute a protocol run in conversation mode. + + Deliberately *not* a turn scheduler. It starts exactly one agent -- the one + the user addressed -- and consultation is driven from inside that run by the + messenger, as a nested call. That is what makes the single-agent loop + unchanged: an agent finishes by writing its answer, not by choosing a + ``finish`` action, and the only new thing in its world is that a peer exists + and can be asked. + + So what remains here is: seed the transcript, run the entry agent, map its + outcome to a terminal conversation state, checkpoint. + """ + async with get_session() as db: + run = await get_protocol_run(db, protocol_run_id) + if run is None: + return + protocol = await get_protocol(db, run.protocol_id) + if protocol is None: + await set_status(db, protocol_run_id, status="failed", error="protocol no longer exists") + return + protocol_id, owner_id, graph = protocol.id, run.owner_id, protocol.graph + if run.protocol_revision_id is not None: + revision = await get_revision(db, run.protocol_revision_id) + if revision is None: + await set_status( + db, protocol_run_id, status="failed", error="published protocol revision no longer exists" + ) + return + graph = revision.graph + experiment_id = protocol.experiment_id + experiment = await get_experiment(db, experiment_id) if experiment_id else None + evaluation_metrics = (experiment.design_spec or {}).get("metrics") if experiment is not None else None + + node = next((n for n in graph.get("nodes") or [] if str(n.get("id")) == entry_agent_id), None) + if node is None or node.get("type") != "agent": + async with get_session() as db: + await set_status(db, protocol_run_id, status="failed", error="entry agent is not an Agent node") + return + + workspace_id = _compute_workspace_id(experiment_id, None, protocol_run_id) + messenger = AgentMessenger( + protocol_id=protocol_id, + protocol_run_id=protocol_run_id, + owner_id=owner_id, + graph=graph, + entry_agent_id=entry_agent_id, + workspace_id=workspace_id, + ) + messenger.append( + from_agent_id=USER_PARTICIPANT, + to_agent_id=entry_agent_id, + parts=[{"kind": "text", "text": user_input}], + ) + + async with get_session() as db: + await set_status(db, protocol_run_id, status="running") + await update_node_run(db, protocol_run_id, entry_agent_id, {"status": "running"}) + await messenger.checkpoint() + + output_text, error, run_id = await _run_agent_node( + node, + protocol_id=protocol_id, + protocol_run_id=protocol_run_id, + owner_id=owner_id, + user_input=user_input, + graph=graph, + workspace_id=workspace_id, + ambient_meta=_ambient_meta_for(graph, entry_agent_id, workspace_id), + evaluation_metrics=evaluation_metrics, + available_agents=await resolve_available_agents(graph, entry_agent_id, owner_id=owner_id), + agent_messenger=messenger, + ) + + cancelled = error == _AGENT_CANCELLED + if not cancelled and error is None: + messenger.append( + from_agent_id=entry_agent_id, + to_agent_id=USER_PARTICIPANT, + parts=[{"kind": "text", "text": output_text or ""}], + state="completed", + ) + # A cap that was hit but absorbed still produced a real answer, so the + # conversation is `completed` -- `limit_reached` is reserved for a cap that + # actually stopped it. Invariant 7 in the run's own status. + if cancelled: + state, status = "canceled", "cancelled" + elif error is not None: + state, status = ("limit_reached", "limit_reached") if messenger.limit_reached else ("failed", "failed") + else: + state, status = "completed", "completed" + messenger.set_state(state) + await messenger.checkpoint() + + async with get_session() as db: + await update_node_run( + db, + protocol_run_id, + entry_agent_id, + { + "status": "cancelled" if cancelled else ("failed" if error else "completed"), + "output_text": output_text, + "error": None if cancelled else error, + "run_id": str(run_id) if run_id else None, + }, + ) + await set_status(db, protocol_run_id, status=status, error=None if cancelled else error) + + +__all__ = [ + "USER_PARTICIPANT", + "AgentMessenger", + "run_conversation", +] diff --git a/src/asaree/services/deadline.py b/src/asaree/services/deadline.py new file mode 100644 index 0000000..975ec4a --- /dev/null +++ b/src/asaree/services/deadline.py @@ -0,0 +1,118 @@ +"""A wall-clock budget for one agent run that stops ticking while it waits. + +``asyncio.wait_for`` takes a fixed duration decided before the wait starts, +which is the wrong shape once an agent can consult a peer: the peer's run +executes *inside* the caller's timeout, so a peer that thinks for three minutes +spends three minutes of a budget meant to bound the caller's own thinking. The +caller would then time out through no fault of its own, and the more useful the +delegation the more likely that becomes. + +**Each agent keeps its own wall-clock allowance.** A :class:`Deadline` is a +mutable expiry that the run's supervisor re-reads while it waits, and +:func:`deadlines_paused` freezes it for the span of a consultation. + +Pausing rather than crediting the time back afterwards is the whole point: a +consultation longer than the caller's remaining budget would blow the deadline +*while it was still in flight*, and no amount of repayment afterwards brings a +run back. The clock has to stop at the moment the caller starts waiting. + +The active chain is a :class:`~contextvars.ContextVar` rather than an argument +threaded through the executor, because the pause has to reach every caller above +the peer, not just the immediate one. A depth-2 consultation blocks the depth-1 +caller *and* the entry agent, and those frames are separated by the whole Motoro +engine -- code that must not learn what a peer is. An asyncio task copies the +context at creation and the tuple holds the same ``Deadline`` objects by +reference, so a nested run's pause is visible to every ancestor still waiting. + +This is deliberately *not* the backstop. The conversation-level wall clock in +:mod:`asaree.services.agent_messenger` never pauses and does count peer time: +total work stays bounded even though no individual agent is charged for +delegating. +""" + +from __future__ import annotations + +import time +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar + +#: What a paused deadline reports as remaining. Deliberately not the frozen +#: remainder, which can be arbitrarily small -- the supervisor uses this value +#: as its wait interval, and a near-zero one would spin. +_PAUSED_POLL_SECONDS = 0.5 + + +class Deadline: + """A point in time a run must finish by, which can be paused. + + Monotonic on purpose: a clock adjustment mid-run must not shorten or + lengthen an agent's allowance. Reentrant, because nested consultations pause + the same ancestor frames more than once and only the outermost resume may + restart the clock. + """ + + __slots__ = ("_expires_at", "_paused_at", "_pauses") + + def __init__(self, seconds: float) -> None: + self._expires_at = time.monotonic() + seconds + self._paused_at: float | None = None + self._pauses = 0 + + def pause(self) -> None: + self._pauses += 1 + if self._pauses == 1: + self._paused_at = time.monotonic() + + def resume(self) -> None: + self._pauses -= 1 + if self._pauses == 0 and self._paused_at is not None: + self._expires_at += time.monotonic() - self._paused_at + self._paused_at = None + + def remaining(self) -> float: + if self._paused_at is not None: + return _PAUSED_POLL_SECONDS + return self._expires_at - time.monotonic() + + def expired(self) -> bool: + return self._paused_at is None and self._expires_at <= time.monotonic() + + +_ACTIVE: ContextVar[tuple[Deadline, ...]] = ContextVar("asaree_active_deadlines", default=()) + + +@contextmanager +def active_deadline(deadline: Deadline) -> Iterator[Deadline]: + """Register *deadline* as the innermost frame of the active caller chain. + + Enter this *before* creating the task that does the work, so the task's + copied context already contains the frame. + """ + token = _ACTIVE.set((*_ACTIVE.get(), deadline)) + try: + yield deadline + finally: + _ACTIVE.reset(token) + + +@contextmanager +def deadlines_paused() -> Iterator[None]: + """Stop the clock on every agent currently waiting, for this block's span. + + Every frame, not just the innermost: a nested consultation blocks its own + caller and everyone above it for the same wall-clock span, so none of them + should be charged for it. The frames are captured on entry so an exception + inside cannot leave a deadline paused forever. + """ + frames = _ACTIVE.get() + for deadline in frames: + deadline.pause() + try: + yield + finally: + for deadline in frames: + deadline.resume() + + +__all__ = ["Deadline", "active_deadline", "deadlines_paused"] diff --git a/src/asaree/services/protocol_execution.py b/src/asaree/services/protocol_execution.py index 7b717ea..57199e1 100644 --- a/src/asaree/services/protocol_execution.py +++ b/src/asaree/services/protocol_execution.py @@ -36,6 +36,7 @@ from motoro.schemas.output import parse_envelope from motoro.schemas.pattern import PatternConfig from motoro.services.mcp_service import hydrate_registry +from motoro.services.skill_service import resolve_skills from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -43,12 +44,14 @@ from asaree.config import get_settings from asaree.models.database import get_session from asaree.models.protocol_run import ProtocolRun +from asaree.services.agent_cards import AgentCard, build_agent_card from asaree.services.dataset_workspaces import ( WorkspaceSeedError, fetch_owned_registration, head_data_locator, seed_cell_workspace, ) +from asaree.services.deadline import Deadline, active_deadline from asaree.services.design_generation import get_design_impact from asaree.services.experiments import get_experiment from asaree.services.factor_bindings import validate_factor_bindings @@ -1164,6 +1167,122 @@ def _resolve_skill_config(graph: dict[str, Any], node_id: str) -> dict[str, Any] return {"skill_ids": skill_ids} if skill_ids else {} +def _is_peer_edge(edge: dict[str, Any], nodes: dict[str, dict[str, Any]]) -> bool: + """A main (untyped) edge joining two Agent nodes. + + This is the *same* edge a normal pipeline run walks to pass one agent's + output to the next -- it is both things, and which one it means is decided + by the run mode, not by the edge. A conversation reads it as an undirected + "these two may consult each other"; ``run_protocol`` keeps reading it as a + directed data-flow. Nothing about the edge is rewritten to say so, which is + why a canvas built for a pipeline run needs no migration to support + conversation. + + Connector-typed edges (``_CONNECTOR_HANDLES``) are configuration rather than + topology, and an Agent<->Critic Gate edge keeps its pipeline meaning, so + neither makes a peer. + """ + if edge.get("targetHandle") in _CONNECTOR_HANDLES: + return False + source = nodes.get(str(edge.get("source"))) + target = nodes.get(str(edge.get("target"))) + return source is not None and target is not None and source.get("type") == target.get("type") == "agent" + + +def _connected_agent_ids(graph: dict[str, Any], node_id: str) -> list[str]: + """Agent node ids reachable from *node_id* over a peer edge, either way. + + Undirected on purpose: an edge's stored source/target records how the user + happened to draw it, not who is allowed to speak. Order is canvas wiring + order, and each peer appears once however many edges join the pair. + """ + nodes = {str(n.get("id")): n for n in graph.get("nodes") or [] if n.get("id")} + peers: list[str] = [] + for edge in graph.get("edges") or []: + if not _is_peer_edge(edge, nodes): + continue + source, target = str(edge.get("source")), str(edge.get("target")) + if source == node_id: + other = target + elif target == node_id: + other = source + else: + continue + if other != node_id and other not in peers: + peers.append(other) + return peers + + +def _can_deliver_communication(graph: dict[str, Any], from_agent_id: str, to_agent_id: str) -> bool: + """Live authorization check, re-run for every consultation. + + Capability is snapshotted, reachability is live: the cards an agent carries + come from the run's published revision, but whether it may still *reach* a + peer is answered against the draft ``Protocol.graph`` at call time. Pulling + the edge on the canvas stops the next consultation mid-run, which is the + behaviour a user unplugging a wire expects. + """ + if from_agent_id == to_agent_id: + return False + return to_agent_id in _connected_agent_ids(graph, from_agent_id) + + +async def resolve_agent_card( + graph: dict[str, Any], + node_id: str, + *, + owner_id: uuid.UUID, + metadata: dict[str, Any] | None = None, +) -> AgentCard | None: + """Project one Agent node into the card its peers see. + + ``None`` for a node that is missing or is not an Agent -- invariant 4: a + non-agent node is never addressable. + + Reads the same graph the run executes from, so a card can never describe an + agent differently from how that agent is about to be configured. + """ + nodes = {str(n.get("id")): n for n in graph.get("nodes") or [] if n.get("id")} + node = nodes.get(node_id) + if node is None or node.get("type") != "agent": + return None + config = (node.get("data") or {}).get("config") or {} + # The registered skill documents, not the ids: a peer reads names and + # descriptions to decide whether this agent is worth asking. Bodies are + # never projected -- progressive disclosure is the owning agent's business. + skills = await resolve_skills(_resolve_skill_config(graph, node_id), owner_id=owner_id) + return build_agent_card( + node_id=node_id, + label=(node.get("data") or {}).get("label"), + description=config.get("description") or "", + goal=config.get("goal") or "", + skills=[dict(s) for s in skills], + model=_resolve_llm_config(graph, node_id).get("model"), + metadata=metadata, + ) + + +async def resolve_available_agents( + graph: dict[str, Any], + node_id: str, + *, + owner_id: uuid.UUID, + metadata: dict[str, Any] | None = None, +) -> list[dict[str, Any]]: + """Serialized cards for every peer connected to *node_id*, in wiring order. + + ``[]`` when nothing is connected -- which is every existing single-agent and + pipeline run, so they carry no new field content, get no prompt section and + are bound no new function schemas (invariant 11). + """ + cards: list[dict[str, Any]] = [] + for peer_id in _connected_agent_ids(graph, node_id): + card = await resolve_agent_card(graph, peer_id, owner_id=owner_id, metadata=metadata) + if card is not None: + cards.append(card.to_dict()) + return cards + + def _resolve_pattern_config(graph: dict[str, Any], node_id: str) -> dict[str, Any]: """``{"execution_pattern": slug, "pattern_params": {slug: {...}}}`` from the node's connected execution-pattern node, or ``{}`` if none is @@ -1581,7 +1700,13 @@ async def _poll_cancel_flag(protocol_run_id: uuid.UUID, cancel_event: asyncio.Ev async def _execute_run_cancellable( - *, run_id: uuid.UUID, protocol_run_id: uuid.UUID, available_tools: list[dict[str, Any]], timeout: float + *, + run_id: uuid.UUID, + protocol_run_id: uuid.UUID, + available_tools: list[dict[str, Any]], + timeout: float, + available_agents: list[dict[str, Any]] | None = None, + agent_messenger: Any = None, ) -> None: """Wraps Motoro's execute_run with the poller above, scoped to exactly this one run's lifetime -- shared by _run_agent_node and @@ -1590,16 +1715,54 @@ async def _execute_run_cancellable( detected, run_protocol's own between-nodes check means no later node ever starts, so nothing else would benefit from a longer-lived poller, and tearing this one down between nodes avoids running it during gaps - where no agent is actually executing.""" + where no agent is actually executing. + + ``available_agents``/``agent_messenger`` travel together and are both empty + for a pipeline run: cards with no messenger would let an agent see peers it + could not reach, and a messenger with no cards would never be called. + + ``timeout`` is enforced through an extendable :class:`Deadline` rather than + ``asyncio.wait_for``, so that time this agent spends *blocked on a peer* can + be handed back to it -- an agent is charged for its own thinking, never for + waiting. With no peers the two are indistinguishable: nothing extends the + deadline, and the run is cancelled at exactly ``timeout`` seconds as + before.""" cancel_event = asyncio.Event() poller = asyncio.create_task(_poll_cancel_flag(protocol_run_id, cancel_event)) try: - await asyncio.wait_for( - execute_run( - run_id=run_id, registry=get_registry(), available_tools=available_tools, cancel_event=cancel_event - ), - timeout=timeout, - ) + # Entered before create_task so the runner's copied context already + # holds this frame, which is how a nested peer run reaches back to + # extend it (see services.deadline). + with active_deadline(Deadline(timeout)) as deadline: + runner = asyncio.create_task( + execute_run( + run_id=run_id, + registry=get_registry(), + available_tools=available_tools, + cancel_event=cancel_event, + available_agents=available_agents or [], + agent_messenger=agent_messenger, + ) + ) + try: + while True: + if deadline.expired(): + raise TimeoutError + # Re-read after every wait rather than waiting once for the + # full span: a consultation may have stopped the clock in + # the meantime, so what looked like the end of the budget + # no longer is. A paged deadline reports a poll interval + # instead of its frozen remainder, which is what makes this + # loop again rather than spin. + done, _ = await asyncio.wait({runner}, timeout=deadline.remaining()) + if done: + await runner # re-raise whatever the run itself raised + return + finally: + if not runner.done(): + runner.cancel() + with contextlib.suppress(asyncio.CancelledError): + await runner finally: poller.cancel() with contextlib.suppress(asyncio.CancelledError): @@ -1612,6 +1775,17 @@ async def _sync_durable_agent(*, name: str, owner_id: uuid.UUID, fields: dict[st Multiple protocol runs can reach the same canvas node concurrently. The durable Motoro Agent is keyed by owner/name, so a losing create retries as an update of the row the concurrent request just inserted. + + **Never put per-run state in ``fields``.** Every entry here must come from + the node's canvas config or its wired connectors -- i.e. from the protocol + itself, which is what this row is a projection of. Run-scoped configuration + belongs on the run: ``create_run`` already takes ``model_config_overrides`` + and ``config_snapshot``. The rule needs stating because this write path + makes the mistake cheap to make and expensive to notice: ``update_agent`` + reads ``None`` as "leave unchanged", so a field written once during one run + is never cleared by any later run and quietly contaminates every future one + (see ``scripts/repair_contaminated_agents.py``, which cleans up an earlier + design that injected a conversation contract this way). """ existing = await get_agent_by_name(name, owner_id=owner_id) if existing is not None: @@ -1636,6 +1810,8 @@ async def _run_agent_node( workspace_id: str | None = None, ambient_meta: dict[str, Any] | None = None, evaluation_metrics: Any = None, + available_agents: list[dict[str, Any]] | None = None, + agent_messenger: Any = None, ) -> tuple[str | None, str | None, uuid.UUID | None]: """Create-or-sync the real agent and run it to completion. Returns ``(output_text, error, run_id)`` -- exactly one of output_text/error is @@ -1643,7 +1819,16 @@ async def _run_agent_node( populated once ``create_run`` succeeds (even on a later timeout/error), since that's what the canvas's Output tab uses to fetch this node's own step trace (``GET /runs/{run_id}/steps``); only ``None`` if agent - creation/sync itself failed before a run could even be created.""" + creation/sync itself failed before a run could even be created. + + ``available_agents`` are the serialized peer cards this node may consult + (:func:`resolve_available_agents`) and ``agent_messenger`` is how a chosen + consultation is delivered. Both default to empty, which is what a pipeline + run passes: peers are a conversation-mode capability, so an ordinary run is + byte-for-byte what it was before. They are deliberately *not* folded into + ``_sync_durable_agent``'s fields -- the card is per-run and derived, and + writing it onto the durable agent row is exactly the contamination this + design avoids.""" config = node["data"]["config"] # Deterministic, not config["name"]: Agent.name is unique per OWNER, not # per protocol, so trusting the freeform (often identically-defaulted) @@ -1740,7 +1925,12 @@ async def _run_agent_node( timeout = agent.max_run_duration_seconds or get_settings().worker_job_timeout_seconds try: await _execute_run_cancellable( - run_id=run.id, protocol_run_id=protocol_run_id, available_tools=gather_tools(agent), timeout=timeout + run_id=run.id, + protocol_run_id=protocol_run_id, + available_tools=gather_tools(agent), + timeout=timeout, + available_agents=available_agents, + agent_messenger=agent_messenger, ) except TimeoutError: return None, f"run exceeded its {timeout}s execution budget", run.id @@ -2490,6 +2680,48 @@ def validate_single_node_runnable(graph: dict[str, Any], node_id: str) -> dict[s return node +def validate_conversation_entry(graph: dict[str, Any], node_id: str) -> dict[str, Any]: + """Validates an agent can host a conversation. Returns the node dict. + + Scoped to the entry agent and its peers, deliberately *not* to the whole + graph: a conversation is a cluster of connected agents, and a half-configured + node in some unrelated corner of the same canvas has nothing to do with it. + + Notably absent: any check on how many peer edges the *graph* has. An earlier + design required exactly one, which made a third agent on the canvas an error + rather than a third participant. + """ + nodes: dict[str, dict[str, Any]] = {str(n["id"]): n for n in graph.get("nodes") or [] if n.get("id")} + node = nodes.get(node_id) + if node is None: + raise ProtocolValidationError(f"No such node: {node_id!r}") + if node.get("type") != "agent": + raise ProtocolValidationError("Only Agent nodes can start a conversation.") + + peers = _connected_agent_ids(graph, node_id) + if not peers: + raise ProtocolValidationError( + f"{_node_display_name(node)!r} isn't connected to another agent, so it has nobody to talk to. " + "Draw an edge between two Agent nodes first." + ) + # The entry agent and every peer it may consult: each needs its own model, + # or the consultation fails partway through a run the user already paid for. + for participant_id in [node_id, *peers]: + participant = nodes[participant_id] + llm_edges = _edges_with_handle(graph, participant_id, "ai", direction="incoming") + if len(llm_edges) != 1: + raise ProtocolValidationError( + f"Node {_node_display_name(participant)!r} must have exactly one AI connection " + f"(found {len(llm_edges)})." + ) + llm_source = nodes.get(str(llm_edges[0]["source"])) + if llm_source is None or llm_source.get("type") not in _LLM_NODE_TYPES: + raise ProtocolValidationError( + f"Node {_node_display_name(participant)!r}'s AI connection must come from an AI node." + ) + return node + + async def _run_single_node( protocol_run_id: uuid.UUID, *, diff --git a/src/asaree/services/protocol_runs.py b/src/asaree/services/protocol_runs.py index 7f30a36..943cb76 100644 --- a/src/asaree/services/protocol_runs.py +++ b/src/asaree/services/protocol_runs.py @@ -23,7 +23,11 @@ from asaree.models.protocol_run import ProtocolRun from asaree.services.factorial_cells import list_replicates -_TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled"}) +# "limit_reached" is terminal too: a conversation that spent its consultation +# budget and could not then produce an answer is finished, not broken, and +# calling it "failed" would hide the one thing a user needs to know to fix it +# (see services.agent_messenger's budget constants). +_TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled", "limit_reached"}) async def create_protocol_run( @@ -199,6 +203,27 @@ async def update_node_run( return run +async def update_conversation( + db: AsyncSession, protocol_run_id: uuid.UUID, conversation: dict[str, Any] +) -> ProtocolRun | None: + """Checkpoint the whole transcript as one document. + + Assigned whole rather than merged: the messenger holds the authoritative + in-memory copy for the life of the run and appends to it, so a partial + merge here could only ever reorder what it already knows. Called before a + peer is allowed to execute and again after it replies, which is what makes + a worker retry able to see exactly how far the conversation got. + """ + run = await get_protocol_run(db, protocol_run_id) + if run is None: + return None + run.conversation = conversation + run.last_heartbeat_at = datetime.now(UTC) + await db.flush() + await db.refresh(run) + return run + + async def update_attempt_result( db: AsyncSession, protocol_run_id: uuid.UUID, *, fields: dict[str, Any] ) -> ProtocolRun | None: diff --git a/tests/test_agent_cards.py b/tests/test_agent_cards.py new file mode 100644 index 0000000..1095ff6 --- /dev/null +++ b/tests/test_agent_cards.py @@ -0,0 +1,236 @@ +"""Peer topology and AgentCard resolution. + +Two properties carry most of the weight here. *Capability is snapshotted, +reachability is live*: a card describes an agent as the run's graph configures +it, while ``_can_deliver_communication`` is re-asked of the draft graph on every +consultation. And *a card is derived, never stored*: every assertion below reads +the graph, because that is the only place the answer lives. +""" + +from __future__ import annotations + +import uuid +from typing import Any + +import pytest + +from asaree.services import protocol_execution as pe +from asaree.services.agent_cards import AgentCard, build_agent_card + +OWNER = uuid.uuid4() + + +def _agent(node_id: str, *, label: str = "", config: dict[str, Any] | None = None) -> dict[str, Any]: + return {"id": node_id, "type": "agent", "data": {"label": label, "config": config or {}}} + + +def _edge(source: str, target: str, handle: str | None = None) -> dict[str, Any]: + edge: dict[str, Any] = {"id": f"{source}-{target}", "source": source, "target": target} + if handle is not None: + edge["targetHandle"] = handle + return edge + + +def _pair_graph() -> dict[str, Any]: + """Planner <-> Critic, joined by one plain edge.""" + return { + "nodes": [ + _agent("planner", label="Planner", config={"description": "Breaks a goal into ordered steps."}), + _agent("critic", label="Critic", config={"description": "Names the weakest claim in a draft."}), + ], + "edges": [_edge("planner", "critic")], + } + + +# ---------------------------------------------------------------------- +# Topology +# ---------------------------------------------------------------------- + + +def test_a_plain_agent_edge_is_a_peer_edge_in_both_directions() -> None: + """The stored source/target records how the user drew the edge, not who + may speak.""" + graph = _pair_graph() + assert pe._connected_agent_ids(graph, "planner") == ["critic"] + assert pe._connected_agent_ids(graph, "critic") == ["planner"] + + +def test_connector_edges_are_configuration_not_topology() -> None: + graph = { + "nodes": [_agent("planner"), _agent("critic")], + # Nonsensical wiring on purpose: even if a connector-typed edge joined + # two Agent nodes, a connector declares capability, not who to talk to. + "edges": [_edge("planner", "critic", "tool")], + } + assert pe._connected_agent_ids(graph, "planner") == [] + + +def test_non_agent_nodes_are_never_peers() -> None: + """Invariant 4 — a Critic Gate edge keeps its pipeline meaning, and an LLM + node is not addressable however it is wired.""" + graph = { + "nodes": [ + _agent("planner"), + {"id": "gate", "type": "critic_gate", "data": {"config": {}}}, + {"id": "llm", "type": "llm_anthropic", "data": {"config": {}}}, + ], + "edges": [_edge("planner", "gate"), _edge("llm", "planner")], + } + assert pe._connected_agent_ids(graph, "planner") == [] + + +def test_peers_are_listed_once_in_wiring_order() -> None: + graph = { + "nodes": [_agent("a"), _agent("b"), _agent("c")], + "edges": [_edge("a", "c"), _edge("b", "a"), {"id": "dup", "source": "c", "target": "a"}], + } + assert pe._connected_agent_ids(graph, "a") == ["c", "b"] + + +def test_an_agent_is_never_its_own_peer() -> None: + graph = {"nodes": [_agent("a")], "edges": [_edge("a", "a")]} + assert pe._connected_agent_ids(graph, "a") == [] + assert pe._can_deliver_communication(graph, "a", "a") is False + + +def test_delivery_is_authorized_by_the_graph_it_is_asked_of() -> None: + """Reachability is live: unplugging the edge on the canvas stops the next + consultation, even though the run's cards came from a revision.""" + connected = _pair_graph() + assert pe._can_deliver_communication(connected, "planner", "critic") is True + assert pe._can_deliver_communication(connected, "critic", "planner") is True + + unplugged = {"nodes": connected["nodes"], "edges": []} + assert pe._can_deliver_communication(unplugged, "planner", "critic") is False + + +def test_an_unconnected_third_agent_is_not_reachable() -> None: + """Invariant 3 — the two clusters on a canvas stay separate.""" + graph = _pair_graph() + graph["nodes"].append(_agent("loner", label="Loner")) + assert pe._connected_agent_ids(graph, "planner") == ["critic"] + assert pe._can_deliver_communication(graph, "planner", "loner") is False + + +# ---------------------------------------------------------------------- +# The card +# ---------------------------------------------------------------------- + + +def test_description_falls_back_to_goal_then_to_a_default() -> None: + """The description becomes the consultation function's description, which + is what a peer's model reads to decide. An empty one is worse than a + generic one.""" + assert ( + build_agent_card( + node_id="n1", label="Critic", description="", goal="Find the weakest claim.", skills=[], model=None + ).description + == "Find the weakest claim." + ) + assert ( + build_agent_card(node_id="n1", label="Critic", description="", goal="", skills=[], model=None).description + == "An agent named Critic." + ) + + +def test_an_unlabeled_agent_falls_back_to_its_node_id() -> None: + assert build_agent_card(node_id="n1", label=None, description="d", goal="", skills=[], model=None).name == "n1" + + +def test_the_card_serializes_to_the_shape_core_consumes() -> None: + card = build_agent_card( + node_id="n1", + label="Critic", + description="Names the weakest claim.", + goal="", + skills=[{"name": "Fact Check", "description": "Verifies claims.", "body": "ignored", "files": {}}], + model="claude-sonnet-5", + metadata={"protocol_id": "p1"}, + ) + assert card.to_dict() == { + "agent_id": "n1", + "name": "Critic", + "description": "Names the weakest claim.", + "skills": [{"id": "fact-check", "name": "Fact Check", "description": "Verifies claims."}], + "metadata": {"protocol_id": "p1", "model": "claude-sonnet-5"}, + } + + +def test_the_card_never_carries_tools_or_the_system_prompt() -> None: + """Exposing another agent's servers invites asking it to proxy a tool call; + the system prompt is user-authored content meaningless to a peer.""" + card = build_agent_card(node_id="n1", label="X", description="d", goal="", skills=[], model=None) + assert set(card.to_dict()) == {"agent_id", "name", "description", "skills", "metadata"} + + +async def test_resolve_agent_card_reads_the_node_it_is_given() -> None: + card = await pe.resolve_agent_card(_pair_graph(), "critic", owner_id=OWNER) + assert isinstance(card, AgentCard) + assert (card.agent_id, card.name) == ("critic", "Critic") + assert card.description == "Names the weakest claim in a draft." + + +async def test_the_card_carries_the_node_id_not_the_motoro_agent_id() -> None: + """The Motoro Agent uuid is an implementation detail of one execution; the + node id is what topology authorizes against and what the transcript + stores.""" + card = await pe.resolve_agent_card(_pair_graph(), "planner", owner_id=OWNER) + assert card is not None + assert card.agent_id == "planner" + + +async def test_the_model_family_comes_from_the_wired_ai_connector() -> None: + graph = _pair_graph() + graph["nodes"].append({"id": "llm", "type": "llm_anthropic", "data": {"config": {"model": "claude-sonnet-5"}}}) + graph["edges"].append(_edge("llm", "critic", "ai")) + + card = await pe.resolve_agent_card(graph, "critic", owner_id=OWNER) + assert card is not None + assert card.metadata["model"] == "claude-sonnet-5" + + +async def test_a_non_agent_node_has_no_card() -> None: + graph = {"nodes": [{"id": "llm", "type": "llm_anthropic", "data": {"config": {}}}], "edges": []} + assert await pe.resolve_agent_card(graph, "llm", owner_id=OWNER) is None + assert await pe.resolve_agent_card(graph, "missing", owner_id=OWNER) is None + + +async def test_wired_skills_reach_the_card_as_names_and_descriptions(monkeypatch: pytest.MonkeyPatch) -> None: + """Names and descriptions only — a skill's body is the owning agent's + business, and a peer cannot load it.""" + skill_id = str(uuid.uuid4()) + graph = _pair_graph() + graph["nodes"].append({"id": "skill1", "type": "skill", "data": {"config": {"skill_id": skill_id}}}) + graph["edges"].append(_edge("skill1", "critic", "skill")) + + seen: dict[str, Any] = {} + + async def _fake_resolve_skills(skill_config: dict[str, Any] | None, **kwargs: Any) -> list[dict[str, Any]]: + seen.update(skill_config or {}) + return [{"name": "Fact Check", "description": "Verifies claims.", "body": "SECRET", "files": {}}] + + monkeypatch.setattr(pe, "resolve_skills", _fake_resolve_skills) + + card = await pe.resolve_agent_card(graph, "critic", owner_id=OWNER) + assert card is not None + assert seen == {"skill_ids": [skill_id]} + assert [s.name for s in card.skills] == ["Fact Check"] + assert "SECRET" not in str(card.to_dict()) + + +async def test_available_agents_are_the_connected_peers_cards() -> None: + agents = await pe.resolve_available_agents(_pair_graph(), "planner", owner_id=OWNER) + assert [a["agent_id"] for a in agents] == ["critic"] + assert agents[0]["name"] == "Critic" + + +async def test_an_agent_with_no_peers_gets_an_empty_list() -> None: + """Invariant 11 — a pipeline or single-agent run carries no new field + content, so nothing about it changes.""" + graph = {"nodes": [_agent("solo", label="Solo")], "edges": []} + assert await pe.resolve_available_agents(graph, "solo", owner_id=OWNER) == [] + + +async def test_a_peer_never_sees_itself_in_its_own_roster() -> None: + agents = await pe.resolve_available_agents(_pair_graph(), "critic", owner_id=OWNER) + assert [a["agent_id"] for a in agents] == ["planner"] diff --git a/tests/test_agent_messenger.py b/tests/test_agent_messenger.py new file mode 100644 index 0000000..51933da --- /dev/null +++ b/tests/test_agent_messenger.py @@ -0,0 +1,404 @@ +"""Consultation delivery: authorization, budgets, transcript, and the clock. + +The property under test throughout is that **a refusal is an answer, not an +error**. Every cap and every authorization failure below asserts on the returned +:class:`AgentReply`, because the whole point is that the calling model reads the +outcome and still writes something useful. A test here that expected an +exception would be asserting the opposite of the design. + +DB-free by construction: ``get_session`` and the five service calls the +messenger makes through it are stubbed, so what runs is exactly the ordering, +budget and authorization logic and nothing else. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import uuid +from collections.abc import AsyncIterator +from typing import Any + +import pytest + +from asaree.services import agent_messenger as am +from asaree.services import protocol_execution as pe +from asaree.services.agent_messenger import AgentMessenger +from asaree.services.deadline import Deadline, active_deadline, deadlines_paused + +PROTOCOL_ID = uuid.uuid4() +RUN_ID = uuid.uuid4() +OWNER = uuid.uuid4() + + +def _agent(node_id: str, label: str) -> dict[str, Any]: + return {"id": node_id, "type": "agent", "data": {"label": label, "config": {"description": label}}} + + +def _graph() -> dict[str, Any]: + """Planner <-> Critic, plus an unconnected Loner.""" + return { + "nodes": [_agent("planner", "Planner"), _agent("critic", "Critic"), _agent("loner", "Loner")], + "edges": [{"id": "e1", "source": "planner", "target": "critic"}], + } + + +class _Row: + def __init__(self, **kwargs: Any) -> None: + self.__dict__.update(kwargs) + + +@pytest.fixture +def stubs(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: + """Cut every DB call, keeping the real authorization and budget logic.""" + state: dict[str, Any] = { + "checkpoints": [], + "node_runs": [], + "cancel_requested_at": None, + "live_graph": _graph(), + "peer_runs": [], + # A tuple, or a callable taking the _run_agent_node kwargs and + # returning one -- the callable form is how a test asserts on state + # mid-consultation. + "peer_result": ("A peer answer.", None, uuid.uuid4()), + } + + @contextlib.asynccontextmanager + async def _session() -> AsyncIterator[None]: + yield None + + async def _update_conversation(_db: Any, _run_id: uuid.UUID, conversation: dict[str, Any]) -> None: + state["checkpoints"].append(conversation) + + async def _get_protocol_run(_db: Any, _run_id: uuid.UUID) -> Any: + return _Row(cancel_requested_at=state["cancel_requested_at"]) + + async def _get_protocol(_db: Any, _protocol_id: uuid.UUID) -> Any: + return _Row(graph=state["live_graph"]) + + async def _update_node_run(_db: Any, _run_id: uuid.UUID, node_id: str, patch: dict[str, Any]) -> None: + state["node_runs"].append((node_id, patch)) + + async def _run_agent_node(node: dict[str, Any], **kwargs: Any) -> tuple[str | None, str | None, uuid.UUID | None]: + state["peer_runs"].append((node["id"], kwargs["user_input"], kwargs["agent_messenger"])) + result = state["peer_result"] + resolved: tuple[str | None, str | None, uuid.UUID | None] = result(kwargs) if callable(result) else result + return resolved + + async def _resolve_available_agents(*_args: Any, **_kwargs: Any) -> list[dict[str, Any]]: + return [] + + monkeypatch.setattr(am, "get_session", _session) + monkeypatch.setattr(am, "update_conversation", _update_conversation) + monkeypatch.setattr(am, "get_protocol_run", _get_protocol_run) + monkeypatch.setattr(am, "get_protocol", _get_protocol) + monkeypatch.setattr(am, "update_node_run", _update_node_run) + monkeypatch.setattr(am, "_run_agent_node", _run_agent_node) + monkeypatch.setattr(am, "resolve_available_agents", _resolve_available_agents) + monkeypatch.setattr(am, "_ambient_meta_for", lambda *a, **k: {}) + return state + + +def _messenger(**kwargs: Any) -> AgentMessenger: + return AgentMessenger( + protocol_id=PROTOCOL_ID, + protocol_run_id=RUN_ID, + owner_id=OWNER, + graph=_graph(), + entry_agent_id="planner", + **kwargs, + ) + + +async def _ask(messenger: AgentMessenger, to: str = "critic", text: str = "What is weak here?") -> Any: + return await messenger.send( + from_agent_id="planner", to_agent_id=to, parts=[{"kind": "text", "text": text}], context=None + ) + + +# ---------------------------------------------------------------------- +# Delivery +# ---------------------------------------------------------------------- + + +async def test_a_consultation_runs_the_peer_and_returns_its_words(stubs: dict[str, Any]) -> None: + reply = await _ask(_messenger()) + assert reply.state == "completed" + assert reply.text == "A peer answer." + assert [r[0] for r in stubs["peer_runs"]] == ["critic"] + assert stubs["peer_runs"][0][1] == "What is weak here?" + + +async def test_the_peer_is_handed_the_messenger_so_it_can_consult_back(stubs: dict[str, Any]) -> None: + """Recursion is the mechanism, bounded by depth -- not something bolted on + top of a flat exchange.""" + await _ask(messenger := _messenger()) + assert stubs["peer_runs"][0][2] is messenger + + +async def test_identity_and_ordering_come_from_the_runtime(stubs: dict[str, Any]) -> None: + """Invariant 2 -- the model supplies content only.""" + messenger = _messenger() + await _ask(messenger) + await _ask(messenger, text="And now?") + messages = messenger.conversation["messages"] + assert [m["sequence"] for m in messages] == [1, 2, 3, 4] + assert len({m["message_id"] for m in messages}) == 4 + assert [(m["from_agent_id"], m["to_agent_id"]) for m in messages] == [ + ("planner", "critic"), + ("critic", "planner"), + ("planner", "critic"), + ("critic", "planner"), + ] + + +async def test_the_transcript_is_checkpointed_around_the_peer_turn(stubs: dict[str, Any]) -> None: + """A worker retry has to be able to see how far the conversation got.""" + + def _result(kwargs: Any) -> tuple[str, None, uuid.UUID]: + # Mid-flight: the question must already be durable, the reply must not. + assert len(stubs["checkpoints"][-1]["messages"]) == 1 + return ("A peer answer.", None, uuid.uuid4()) + + stubs["peer_result"] = _result + await _ask(_messenger()) + assert len(stubs["checkpoints"][-1]["messages"]) == 2 + + +async def test_a_consulted_peer_shows_on_the_canvas_as_a_node_that_ran(stubs: dict[str, Any]) -> None: + await _ask(_messenger()) + assert [n for n, _ in stubs["node_runs"]] == ["critic", "critic"] + assert stubs["node_runs"][0][1]["status"] == "running" + assert stubs["node_runs"][1][1]["status"] == "completed" + + +async def test_a_peer_that_fails_is_a_reply_not_an_exception(stubs: dict[str, Any]) -> None: + stubs["peer_result"] = (None, "boom", uuid.uuid4()) + reply = await _ask(_messenger()) + assert reply.state == "failed" + assert reply.error == "boom" + assert "boom" in reply.text + + +async def test_a_cancelled_peer_is_canceled_not_failed(stubs: dict[str, Any]) -> None: + """A Stop is not an error, and reporting it as one would put a red node on + a canvas the user themselves stopped.""" + stubs["peer_result"] = (None, pe._AGENT_CANCELLED, uuid.uuid4()) + reply = await _ask(_messenger()) + assert reply.state == "canceled" + assert reply.error is None + + +async def test_a_silent_peer_still_says_something(stubs: dict[str, Any]) -> None: + """An empty function result reads to a model as a malfunction rather than + as an answer to work around.""" + stubs["peer_result"] = (" ", None, uuid.uuid4()) + reply = await _ask(_messenger()) + assert reply.state == "completed" + assert reply.text + + +# ---------------------------------------------------------------------- +# Authorization +# ---------------------------------------------------------------------- + + +async def test_an_unconnected_agent_cannot_be_consulted(stubs: dict[str, Any]) -> None: + """Invariant 3 -- and it is a rejection, so the caller can still answer.""" + reply = await _ask(_messenger(), to="loner") + assert reply.state == "rejected" + assert "not connected" in reply.text + assert stubs["peer_runs"] == [] + + +async def test_pulling_the_edge_mid_run_stops_the_next_consultation(stubs: dict[str, Any]) -> None: + """Reachability is live: the run's cards came from a revision, but delivery + is re-authorized against the draft graph every single time.""" + messenger = _messenger() + assert (await _ask(messenger)).state == "completed" + stubs["live_graph"] = {"nodes": _graph()["nodes"], "edges": []} + assert (await _ask(messenger)).state == "rejected" + assert len(stubs["peer_runs"]) == 1 + + +async def test_a_refused_consultation_is_still_in_the_transcript(stubs: dict[str, Any]) -> None: + """Invariant 10 -- a silently dropped question is the failure this makes + debuggable.""" + messenger = _messenger() + await _ask(messenger, to="loner") + messages = messenger.conversation["messages"] + assert [m["to_agent_id"] for m in messages] == ["loner", "planner"] + assert messages[1]["state"] == "rejected" + + +async def test_a_stop_between_consultations_prevents_further_ones(stubs: dict[str, Any]) -> None: + """Invariant 8.""" + from datetime import UTC, datetime + + stubs["cancel_requested_at"] = datetime.now(UTC) + reply = await _ask(_messenger()) + assert reply.state == "rejected" + assert "cancelled" in reply.text + assert stubs["peer_runs"] == [] + + +# ---------------------------------------------------------------------- +# Budgets +# ---------------------------------------------------------------------- + + +async def test_the_execution_budget_is_spent_not_bypassed( + stubs: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(am, "_MAX_PEER_EXECUTIONS", 2) + messenger = _messenger() + assert (await _ask(messenger)).state == "completed" + assert (await _ask(messenger)).state == "completed" + exhausted = await _ask(messenger) + assert exhausted.state == "rejected" + assert "consultations" in exhausted.text + assert len(stubs["peer_runs"]) == 2 + assert messenger.limit_reached is True + + +async def test_depth_is_capped_and_the_cap_is_a_reply(stubs: dict[str, Any], monkeypatch: pytest.MonkeyPatch) -> None: + """The consulted peer consults back *from inside its own turn*; at the cap + it gets prose telling it to answer instead. Invariants 6 and 7 together.""" + monkeypatch.setattr(am, "_MAX_CONSULT_DEPTH", 1) + nested_replies: list[Any] = [] + + async def _run_agent_node(node: dict[str, Any], **kwargs: Any) -> tuple[str, None, uuid.UUID]: + messenger: AgentMessenger = kwargs["agent_messenger"] + nested_replies.append( + await messenger.send( + from_agent_id="critic", + to_agent_id="planner", + parts=[{"kind": "text", "text": "Clarify?"}], + context=None, + ) + ) + return ("A peer answer.", None, uuid.uuid4()) + + monkeypatch.setattr(am, "_run_agent_node", _run_agent_node) + messenger = _messenger() + outer = await _ask(messenger) + + assert outer.state == "completed" + assert [r.state for r in nested_replies] == ["rejected"] + assert "nested" in nested_replies[0].text + assert messenger.limit_reached is True + + +async def test_depth_is_released_when_a_consultation_returns( + stubs: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + """A cap on how deep, not on how many in a row -- two sequential + consultations from the entry agent are both depth 1.""" + monkeypatch.setattr(am, "_MAX_CONSULT_DEPTH", 1) + messenger = _messenger() + assert (await _ask(messenger)).state == "completed" + assert (await _ask(messenger)).state == "completed" + + +async def test_the_conversation_wall_clock_is_the_backstop( + stubs: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + """Peer time is given back to individual agents but never to this cap -- + otherwise nothing would bound the total.""" + from datetime import timedelta + + monkeypatch.setattr(am, "_MAX_CONVERSATION_DURATION", timedelta(seconds=0)) + reply = await _ask(_messenger()) + assert reply.state == "rejected" + assert "seconds" in reply.text + assert stubs["peer_runs"] == [] + + +# ---------------------------------------------------------------------- +# Timeout accounting +# ---------------------------------------------------------------------- + + +async def test_a_paused_deadline_does_not_expire_while_it_waits() -> None: + deadline = Deadline(0.05) + deadline.pause() + await asyncio.sleep(0.15) + assert deadline.expired() is False + deadline.resume() + # The paused span was given back, so what was left before the pause is + # still left after it. + assert 0 < deadline.remaining() <= 0.05 + assert deadline.expired() is False + + +async def test_every_caller_in_the_chain_stops_ticking() -> None: + """A depth-2 consultation blocks its own caller and the entry agent for the + same span, so neither is charged for it.""" + entry, middle = Deadline(0.05), Deadline(0.05) + with active_deadline(entry), active_deadline(middle), deadlines_paused(): + await asyncio.sleep(0.15) + assert (entry.expired(), middle.expired()) == (False, False) + + +async def test_a_finished_frame_is_no_longer_paused() -> None: + """Leaving ``active_deadline`` takes the frame out of the chain, so a later + consultation cannot freeze a run that already ended.""" + with active_deadline(Deadline(10)): + with active_deadline(inner := Deadline(0.05)): + pass + with deadlines_paused(): + await asyncio.sleep(0.15) + assert inner.expired() is True + + +async def test_nested_pauses_only_restart_the_clock_once() -> None: + deadline = Deadline(0.05) + with active_deadline(deadline), deadlines_paused(): + with deadlines_paused(): + await asyncio.sleep(0.1) + assert deadline.expired() is False + await asyncio.sleep(0.1) + assert deadline.expired() is False + + +async def test_peer_time_does_not_burn_the_callers_own_budget(monkeypatch: pytest.MonkeyPatch) -> None: + """The decision this feature turns on: an agent is charged for its own + thinking, never for waiting on a peer. A consultation longer than the + caller's whole allowance must not time the caller out.""" + + async def _never_cancels(*_args: Any, **_kwargs: Any) -> None: + await asyncio.Event().wait() + + async def _execute_run(**_kwargs: Any) -> None: + # Stand in for the engine: think briefly, block on a peer for longer + # than the entire budget, then finish. Post-hoc repayment would not + # save this run -- the deadline would already have fired mid-wait -- + # which is why the clock stops instead. + await asyncio.sleep(0.05) + with deadlines_paused(): + await asyncio.sleep(0.3) + await asyncio.sleep(0.05) + + monkeypatch.setattr(pe, "_poll_cancel_flag", _never_cancels) + monkeypatch.setattr(pe, "execute_run", _execute_run) + monkeypatch.setattr(pe, "get_registry", lambda: None) + + await pe._execute_run_cancellable(run_id=uuid.uuid4(), protocol_run_id=RUN_ID, available_tools=[], timeout=0.25) + + +async def test_an_agent_that_overruns_on_its_own_still_times_out(monkeypatch: pytest.MonkeyPatch) -> None: + """The other half: the deadline is still a real limit when nothing extends + it, which is every pipeline run.""" + + async def _never_cancels(*_args: Any, **_kwargs: Any) -> None: + await asyncio.Event().wait() + + async def _execute_run(**_kwargs: Any) -> None: + await asyncio.sleep(5) + + monkeypatch.setattr(pe, "_poll_cancel_flag", _never_cancels) + monkeypatch.setattr(pe, "execute_run", _execute_run) + monkeypatch.setattr(pe, "get_registry", lambda: None) + + with pytest.raises(TimeoutError): + await pe._execute_run_cancellable(run_id=uuid.uuid4(), protocol_run_id=RUN_ID, available_tools=[], timeout=0.1) From 17efb38a538d99905351061162c6ec488f7294db Mon Sep 17 00:00:00 2001 From: Jay Moran Date: Tue, 8 Sep 2026 15:49:38 -0700 Subject: [PATCH 03/57] Add the conversation-mode run endpoint POST /protocols/{id}/conversations addresses one agent on the canvas and lets it consult the peers it is connected to. The request carries only an entry point and an opening question: who else may participate comes from the graph, so a client cannot widen a run's reach by naming extra agents. It creates the same ProtocolRun against the same published revision a pipeline run uses -- what differs is only which worker function walks it. Peers are a capability of the run mode, not a different kind of protocol. execute_conversation_task is a separate arq function rather than a flag on the run row: which mode a run is in is decided once, by the endpoint that started it, and a column would let a retry wander into the wrong mode if it were ever written wrong. Both tasks now share _guarded_protocol_run for the durability guards, none of which care whether the run walks a graph or hosts a conversation. They share the protocol-run: job-id namespace so a run can never have one of each queued against it. validate_conversation_entry rejects an entry that is missing, not an agent, has nobody to talk to, or whose peer has no model -- 422 before a run row exists, rather than a job that dies on the worker. --- src/asaree/api/protocols.py | 42 +++++++++++++++++- src/asaree/worker/enqueue.py | 17 ++++++++ src/asaree/worker/settings.py | 8 +++- src/asaree/worker/tasks.py | 38 +++++++++++++++- tests/test_protocol_execution.py | 74 ++++++++++++++++++++++++++++++-- 5 files changed, 172 insertions(+), 7 deletions(-) diff --git a/src/asaree/api/protocols.py b/src/asaree/api/protocols.py index 8c9b014..c39ddf7 100644 --- a/src/asaree/api/protocols.py +++ b/src/asaree/api/protocols.py @@ -24,6 +24,7 @@ plan_cell_runs, plan_single_replicate_run, topological_order, + validate_conversation_entry, validate_coordination_strategy, validate_single_node_runnable, ) @@ -47,7 +48,7 @@ list_protocols, update_protocol, ) -from asaree.worker.enqueue import enqueue_protocol_run +from asaree.worker.enqueue import enqueue_conversation, enqueue_protocol_run router = APIRouter(prefix="/protocols", tags=["protocols"]) @@ -95,6 +96,9 @@ class ProtocolRunResponse(BaseModel): design_revision_id: uuid.UUID | None protocol_revision_id: uuid.UUID | None target_node_id: str | None + # Null for every pipeline run. Populated once a conversation-mode run's + # agents start talking -- see models/protocol_run.py for the shape. + conversation: dict[str, Any] | None = None cancel_requested_at: datetime | None created_at: datetime updated_at: datetime @@ -122,6 +126,18 @@ class CreateProtocolRunRequest(BaseModel): replicate_label: str | None = None +class StartConversationRequest(BaseModel): + """Address one agent on the canvas and let it consult its connected peers. + + Only an entry point and an opening question: who else may participate comes + from the graph, not from this body, so a client cannot widen a run's reach + by naming extra agents. + """ + + entry_agent_id: str + user_input: str + + class CellRunBatchRequest(BaseModel): """Previously completed replicates the user explicitly chose to run again. @@ -317,6 +333,30 @@ async def create_protocol_run_endpoint( return ProtocolRunResponse.model_validate(run) +@router.post("/{protocol_id}/conversations", response_model=ProtocolRunResponse, status_code=201) +async def start_conversation_endpoint( + protocol_id: uuid.UUID, user: CurrentUser, db: DbSession, body: StartConversationRequest +) -> ProtocolRunResponse: + """Start a conversation-mode run: address one agent, and let it consult the + peers it is connected to. + + The same ``ProtocolRun`` a pipeline run uses, against the same published + revision -- what differs is only which worker function walks it. Peers are a + capability of the run mode, not a different kind of protocol. + """ + protocol = await _get_owned_protocol(db, protocol_id, user) + revision = await _require_published_revision(db, protocol) + try: + validate_conversation_entry(revision.graph, body.entry_agent_id) + except ProtocolValidationError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + run = await create_protocol_run( + db, protocol_id=protocol_id, owner_id=user.id, protocol_revision_id=revision.id + ) + await enqueue_conversation(run.id, entry_agent_id=body.entry_agent_id, user_input=body.user_input) + return ProtocolRunResponse.model_validate(run) + + @router.post("/{protocol_id}/nodes/{node_id}/run", response_model=ProtocolRunResponse, status_code=201) async def run_single_node_endpoint( protocol_id: uuid.UUID, node_id: str, user: CurrentUser, db: DbSession diff --git a/src/asaree/worker/enqueue.py b/src/asaree/worker/enqueue.py index 953fed3..abec7c6 100644 --- a/src/asaree/worker/enqueue.py +++ b/src/asaree/worker/enqueue.py @@ -43,6 +43,23 @@ async def enqueue_protocol_run(protocol_run_id: uuid.UUID) -> None: await pool.enqueue_job("execute_protocol_run_task", str(protocol_run_id), _job_id=f"protocol-run:{protocol_run_id}") +async def enqueue_conversation(protocol_run_id: uuid.UUID, *, entry_agent_id: str, user_input: str) -> None: + """Same idempotent-enqueue pattern, for asaree.worker.tasks.execute_conversation_task. + + Shares the ``protocol-run:`` job-id namespace with :func:`enqueue_protocol_run` + on purpose: both execute the same row, so a run must never be able to have + one of each queued against it. + """ + pool = await _get_pool() + await pool.enqueue_job( + "execute_conversation_task", + str(protocol_run_id), + entry_agent_id, + user_input, + _job_id=f"protocol-run:{protocol_run_id}", + ) + + async def enqueue_metric_evaluation(protocol_run_id: uuid.UUID) -> None: """Queue an idempotent backfill/retry of configured post-run metrics.""" pool = await _get_pool() diff --git a/src/asaree/worker/settings.py b/src/asaree/worker/settings.py index ce21b1b..965d4ce 100644 --- a/src/asaree/worker/settings.py +++ b/src/asaree/worker/settings.py @@ -34,6 +34,7 @@ from asaree.worker.tasks import ( check_stale_runs, evaluate_protocol_run_metrics_task, + execute_conversation_task, execute_protocol_run_task, execute_run_task, ) @@ -62,7 +63,12 @@ async def on_shutdown(ctx: dict[str, Any]) -> None: class WorkerSettings: - functions = [execute_run_task, execute_protocol_run_task, evaluate_protocol_run_metrics_task] + functions = [ + execute_run_task, + execute_protocol_run_task, + execute_conversation_task, + evaluate_protocol_run_metrics_task, + ] cron_jobs = [cron(check_stale_runs, second={0, 30})] redis_settings = RedisSettings.from_dsn(get_settings().redis_url) on_startup = on_startup diff --git a/src/asaree/worker/tasks.py b/src/asaree/worker/tasks.py index 11965fb..a016b93 100644 --- a/src/asaree/worker/tasks.py +++ b/src/asaree/worker/tasks.py @@ -13,7 +13,7 @@ import asyncio import logging import uuid -from collections.abc import Coroutine +from collections.abc import Callable, Coroutine from datetime import UTC, datetime, timedelta from typing import Any @@ -34,6 +34,7 @@ import asaree.models.user # noqa: F401 -- registers users for Protocol/ProtocolRun's owner_id FK from asaree.config import get_settings from asaree.models.database import get_session +from asaree.services.agent_messenger import run_conversation from asaree.services.protocol_execution import evaluate_protocol_run_metrics, run_protocol from asaree.services.protocol_runs import fail_protocol_run, get_protocol_run, list_stale_protocol_runs from asaree.services.run_tools import gather_tools @@ -103,6 +104,39 @@ async def execute_protocol_run_task(ctx: dict[str, Any], protocol_run_id_str: st timeout/exception rather than letting arq retry a partially-executed graph (real agent runs, real tool calls -- not safe to assume idempotent). """ + await _guarded_protocol_run(protocol_run_id_str, run_protocol) + + +async def execute_conversation_task( + ctx: dict[str, Any], protocol_run_id_str: str, entry_agent_id: str, user_input: str +) -> None: + """The conversation-mode counterpart of ``execute_protocol_run_task``. + + A separate arq function rather than a flag on the run row: which mode a run + is in is decided once, by the endpoint that started it, and a row column + would let a retry of a pipeline run wander into conversation mode (or the + reverse) if it were ever written wrong. The entry agent and the opening + question ride as job arguments because nothing else needs them persisted -- + the transcript records them the moment the run starts. + """ + await _guarded_protocol_run( + protocol_run_id_str, + lambda protocol_run_id: run_conversation( + protocol_run_id, entry_agent_id=entry_agent_id, user_input=user_input + ), + ) + + +async def _guarded_protocol_run( + protocol_run_id_str: str, work: Callable[[uuid.UUID], Coroutine[Any, Any, None]] +) -> None: + """The durability guards both protocol-run tasks need, around either body. + + Shared rather than duplicated because every branch below is about the run + *row* -- skip a non-actionable status, force-fail on timeout, record a + cancellation before re-raising -- and none of it knows or cares whether the + run walks a graph or hosts a conversation. + """ protocol_run_id = uuid.UUID(protocol_run_id_str) async with get_session() as db: run = await get_protocol_run(db, protocol_run_id) @@ -118,7 +152,7 @@ async def execute_protocol_run_task(ctx: dict[str, Any], protocol_run_id_str: st timeout = get_settings().worker_job_timeout_seconds try: - await asyncio.wait_for(run_protocol(protocol_run_id), timeout=timeout) + await asyncio.wait_for(work(protocol_run_id), timeout=timeout) except TimeoutError: async with get_session() as db: await fail_protocol_run(db, protocol_run_id, error=f"protocol run exceeded its {timeout}s execution budget") diff --git a/tests/test_protocol_execution.py b/tests/test_protocol_execution.py index 2defc1e..8aa7a44 100644 --- a/tests/test_protocol_execution.py +++ b/tests/test_protocol_execution.py @@ -1554,7 +1554,7 @@ async def fake_run_agent_node(node, *, graph, workspace_id=None, **kwargs): }, ) protocol_id = protocol.id - await upsert_replicate( + replicate = await upsert_replicate( db, experiment_id=experiment_id, replicate_label="only-cell", @@ -1566,6 +1566,11 @@ async def fake_run_agent_node(node, *, graph, workspace_id=None, **kwargs): owner_id=owner_id, replicate_label="only-cell", factor_values={"Temperature": 0.1}, + # Claims the replicate slot, exactly as plan_cell_runs does. Without + # it run_protocol's write-back is correctly skipped: + # is_current_replicate_attempt reads run.replicate_result_id to decide + # whether this run still owns the slot's latest projection. + replicate_result_id=replicate.id, ) run_id = run.id @@ -1613,11 +1618,17 @@ async def _run_single_cell_protocol(owner_id: uuid.UUID) -> tuple[uuid.UUID, str }, ) protocol_id = protocol.id - await upsert_replicate( + replicate = await upsert_replicate( db, experiment_id=experiment_id, replicate_label="only-cell", fields={"factor_values": {}} ) run = await create_protocol_run( - db, protocol_id=protocol_id, owner_id=owner_id, replicate_label="only-cell" + db, + protocol_id=protocol_id, + owner_id=owner_id, + replicate_label="only-cell", + # Claims the replicate slot the way plan_cell_runs does -- see + # is_current_replicate_attempt, which gates run_protocol's write-back. + replicate_result_id=replicate.id, ) run_id = run.id return experiment_id, "only-cell", protocol_id, run_id @@ -3036,6 +3047,63 @@ def test_validate_single_node_runnable_accepts_a_valid_standalone_agent() -> Non assert pe.validate_single_node_runnable(graph, "a") is agent +# --- validate_conversation_entry --------------------------------------------- + + +def _conversation_graph(*agent_ids: str) -> dict: + """Agents in a chain, each with its own LLM, joined by plain main edges.""" + nodes: list[dict] = [_llm_node()] + edges: list[dict] = [] + for agent_id in agent_ids: + agent, llm_edge = _agent_with_llm(agent_id) + nodes.append(agent) + edges.append(llm_edge) + for source, target in zip(agent_ids, agent_ids[1:], strict=False): + edges += _edges((source, target)) + return {"nodes": nodes, "edges": edges} + + +def test_validate_conversation_entry_rejects_a_missing_node() -> None: + with pytest.raises(ProtocolValidationError, match="No such node"): + pe.validate_conversation_entry(_conversation_graph("a", "b"), "nope") + + +def test_validate_conversation_entry_rejects_a_non_agent_node() -> None: + graph = _conversation_graph("a", "b") + graph["nodes"].append(_node("g1", "critic_gate")) + with pytest.raises(ProtocolValidationError, match="Only Agent nodes"): + pe.validate_conversation_entry(graph, "g1") + + +def test_validate_conversation_entry_rejects_an_agent_with_nobody_to_talk_to() -> None: + with pytest.raises(ProtocolValidationError, match="nobody to talk to"): + pe.validate_conversation_entry(_conversation_graph("a"), "a") + + +def test_validate_conversation_entry_rejects_a_peer_with_no_model() -> None: + """A peer's own wiring is checked too: it will really run, and finding out + mid-conversation costs the user a run they already paid for.""" + graph = _conversation_graph("a", "b") + graph["edges"] = [e for e in graph["edges"] if e.get("target") != "b" or e.get("targetHandle") != "ai"] + with pytest.raises(ProtocolValidationError, match="exactly one AI connection"): + pe.validate_conversation_entry(graph, "a") + + +def test_validate_conversation_entry_accepts_a_third_agent_on_the_canvas() -> None: + """Three connected agents are three participants, not an error -- the + validator asks nothing about how many peer edges the graph has.""" + graph = _conversation_graph("a", "b", "c") + assert pe.validate_conversation_entry(graph, "b")["id"] == "b" + + +def test_validate_conversation_entry_ignores_an_unrelated_broken_node() -> None: + """Scoped to the entry agent and its peers: a half-wired node in another + corner of the same canvas has nothing to do with this conversation.""" + graph = _conversation_graph("a", "b") + graph["nodes"].append(_node("stranded", "agent")) + assert pe.validate_conversation_entry(graph, "a")["id"] == "a" + + async def test_run_single_node_ignores_an_unrelated_broken_sibling_node( owner_id: uuid.UUID, monkeypatch: pytest.MonkeyPatch ) -> None: From 656735f04622906a47f52f148fb3bbf1dd6d7119 Mon Sep 17 00:00:00 2001 From: Jay Moran Date: Tue, 8 Sep 2026 15:49:48 -0700 Subject: [PATCH 04/57] Show conversations on the protocol canvas A "Converse" action alongside Run, a live transcript panel over the canvas, and a "can consult" caption on every peer edge. The caption rather than a restyle is the point: a solid Agent-to-Agent edge is BOTH the left-to-right pipeline edge a normal run walks AND the undirected "these two may consult each other" edge a conversation run reads. The run mode picks, so the edge itself must not commit to either. ConversationTranscript renders parts, not a bare string, and is a flat ordered list rather than a threaded tree -- sequence is assigned by the runtime and one agent runs at a time, so the order things happened in IS the structure. limit_reached joins TERMINAL_RUN_STATUSES: a budget-exhausted conversation is finished, not stuck. An agent wired to peers on a model that cannot be sent function schemas would silently never consult them, so AgentNode warns. That needed a new fact on the wire, supports_tool_calling on GET /llm-settings/{provider}/models, and it is bool | None rather than bool: services/tool_calling.py returns None when litellm has never heard of the model. Motoro's model_supports_tool_calling collapses "unknown" into False, which is the right conservative default when the answer picks an execution pattern and the wrong one to show a user -- every Azure Foundry deployment name is unknown, so False would put a warning on every Azure agent. The canvas injects only the wiring facts and AgentNode subscribes to the model list itself, so the warning appears when the list loads. --- frontend/src/api/client.ts | 10 + .../protocol/ConversationTranscript.tsx | 99 ++++++++++ .../components/protocol/ProtocolCanvas.tsx | 182 +++++++++++++++++- .../components/protocol/RunConfirmDialog.tsx | 22 ++- .../protocol/edges/InteractEdge.tsx | 38 +++- .../components/protocol/nodes/AgentNode.tsx | 29 ++- .../src/components/protocol/runSummary.ts | 19 +- frontend/src/lib/protocolRun.ts | 9 +- frontend/src/types/llmSettings.ts | 4 + frontend/src/types/protocols.ts | 31 ++- src/asaree/api/llm_settings.py | 11 ++ src/asaree/services/tool_calling.py | 43 +++++ 12 files changed, 477 insertions(+), 20 deletions(-) create mode 100644 frontend/src/components/protocol/ConversationTranscript.tsx create mode 100644 src/asaree/services/tool_calling.py diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index ef459ec..5d619a0 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -284,6 +284,16 @@ export const protocolsApi = { // runnable Agent (see validate_single_node_runnable). Same polling shape // as a plain run (getRun), just with node_runs carrying only this one key. runNode: (id: string, nodeId: string) => request(`/protocols/${id}/nodes/${nodeId}/run`, { method: 'POST' }), + // Conversation mode: address one agent and let it consult the peers it's + // wired to. Returns the same ProtocolRun a pipeline run does -- poll getRun + // the same way; what's new is `conversation` on the response, the transcript + // as it grows. 422 if the entry agent isn't an Agent node, has no peers, or + // any participant is missing its single LLM connection. + startConversation: (id: string, data: { entryAgentId: string; userInput: string }) => + request(`/protocols/${id}/conversations`, { + method: 'POST', + body: { entry_agent_id: data.entryAgentId, user_input: data.userInput }, + }), getRevision: (id: string, revisionId: string) => request(`/protocols/${id}/revisions/${revisionId}`), getRun: (id: string, runId: string) => request(`/protocols/${id}/runs/${runId}`), // Only raises cancel_requested_at -- a no-op (200, unchanged row) once the diff --git a/frontend/src/components/protocol/ConversationTranscript.tsx b/frontend/src/components/protocol/ConversationTranscript.tsx new file mode 100644 index 0000000..0e069a4 --- /dev/null +++ b/frontend/src/components/protocol/ConversationTranscript.tsx @@ -0,0 +1,99 @@ +import type { Conversation, ConversationMessage } from '@/types/protocols' + +// The literal id the backend uses for the human on both ends of a +// conversation: the opening question comes from "user", and the entry agent's +// final answer goes back to it (services/agent_messenger.py's +// USER_PARTICIPANT). +const USER_PARTICIPANT = 'user' + +// A2A TaskState, as the messenger records it on each reply. Only a reply +// carries one -- a request has no outcome yet -- and only the ones that mean +// something went sideways are worth colouring. +const REPLY_BADGE: Record = { + completed: { label: 'answered', className: 'text-[color:var(--chart-3)]' }, + failed: { label: 'failed', className: 'text-destructive' }, + canceled: { label: 'cancelled', className: 'text-muted-foreground' }, + rejected: { label: 'refused', className: 'text-[color:var(--chart-4)]' }, +} + +const CONVERSATION_STATE_LABEL: Record = { + working: 'Running', + completed: 'Completed', + failed: 'Failed', + canceled: 'Cancelled', + limit_reached: 'Limit reached', +} + +// Parts, not a bare string: the backend already speaks A2A's part shape, so a +// future data or file part is a new branch here rather than a rewrite of every +// caller. Non-text parts are named rather than dropped -- a transcript that +// silently omits half a reply is worse than one that says "1 data part". +function MessageBody({ parts }: { parts: ConversationMessage['parts'] }) { + return ( + <> + {parts.map((part, i) => { + if (part.kind === 'text') { + return ( +

+ {part.text} +

+ ) + } + return ( +

+ [{part.kind} part] +

+ ) + })} + + ) +} + +// The live transcript of a conversation-mode run, over the canvas. Deliberately +// a flat, ordered list rather than a threaded tree: `sequence` is assigned by +// the runtime and one agent runs at a time, so the order things happened in IS +// the structure -- nesting would only re-derive what the reading order already +// shows. +export function ConversationTranscript({ + conversation, + agentNames, +}: { + conversation: Conversation + // Node id -> the label shown on that node's card, so the transcript names + // agents the way the canvas does. Ids that aren't on the canvas (a node + // deleted since the run) fall back to the raw id rather than disappearing. + agentNames: Map +}) { + const nameOf = (id: string) => (id === USER_PARTICIPANT ? 'You' : agentNames.get(id) ?? id) + const stateLabel = CONVERSATION_STATE_LABEL[conversation.state] ?? conversation.state + + return ( + + ) +} diff --git a/frontend/src/components/protocol/ProtocolCanvas.tsx b/frontend/src/components/protocol/ProtocolCanvas.tsx index 4bcf20f..38c208f 100644 --- a/frontend/src/components/protocol/ProtocolCanvas.tsx +++ b/frontend/src/components/protocol/ProtocolCanvas.tsx @@ -16,8 +16,9 @@ import { type Viewport, } from '@xyflow/react' import '@xyflow/react/dist/style.css' -import { Lock, Plus, Square, X } from 'lucide-react' +import { Lock, MessagesSquare, Plus, Square, X } from 'lucide-react' import { Button } from '@/components/ui/button' +import { Textarea } from '@/components/ui/textarea' import { ApiError, experimentsApi, protocolsApi } from '@/api/client' import { newNodeId } from '@/lib/nodeId' import { protocolForExperimentQueryKey, protocolGraphQueryKey, toPersistedGraph } from '@/lib/protocolGraph' @@ -102,6 +103,7 @@ import { OkfDocumentNodeInspector } from './OkfDocumentNodeInspector' import { SkillBrowserPanel } from './SkillBrowserPanel' import { SKILL_BROWSE, nodeDataForSkill } from './skillCatalog' import { SkillNodeInspector } from './SkillNodeInspector' +import { ConversationTranscript } from './ConversationTranscript' import { InteractEdge } from './edges/InteractEdge' import { AgentNode } from './nodes/AgentNode' import { CriticGateNode } from './nodes/CriticGateNode' @@ -472,6 +474,12 @@ export const ProtocolCanvas = forwardRef(null) + // Conversation mode's own confirm state. Separate from pendingRunConfirm + // because the dialog needs two fields filled in (who to ask, and what) before + // the scope it confirms even exists. + const [conversationDialogOpen, setConversationDialogOpen] = useState(false) + const [conversationInput, setConversationInput] = useState('') + const [conversationEntryAgentId, setConversationEntryAgentId] = useState('') const [runId, setRunId] = useState(null) const paneRef = useRef(null) const { screenToFlowPosition, fitView } = useReactFlow() @@ -530,9 +538,16 @@ export const ProtocolCanvas = forwardRef setRunId(run.id), }) + // Conversation mode reuses the same runId/runQuery polling as every other + // run -- what's new is `conversation` on the polled row, which the transcript + // panel reads. + const conversationMutation = useMutation({ + mutationFn: (data: { entryAgentId: string; userInput: string }) => protocolsApi.startConversation(protocolId, data), + onSuccess: (run) => { + setRunId(run.id) + setConversationDialogOpen(false) + setConversationInput('') + }, + }) + // First refetchInterval-based poll in this codebase -- no existing // long-running-job UI to mirror. Function form so polling stops itself // once the run reaches a terminal status, rather than polling forever. @@ -572,7 +599,10 @@ export const ProtocolCanvas = forwardRef { + const nodeTypeById = new Map(nodes.map((n) => [n.id, n.type])) + const map = new Map() + const edgeIds = new Set() + const link = (a: string, b: string) => map.set(a, [...(map.get(a) ?? []), b]) + for (const e of edges) { + if (CONNECTOR_HANDLES.has(e.targetHandle ?? '')) continue + if (nodeTypeById.get(e.source) !== 'agent' || nodeTypeById.get(e.target) !== 'agent') continue + if (e.source === e.target) continue + edgeIds.add(e.id) + link(e.source, e.target) + link(e.target, e.source) + } + return { peerIdsByAgent: map, peerEdgeIds: edgeIds } + }, [nodes, edges]) + + // Stamped onto the RENDERED copy only, never onto the persisted `edges` + // state -- an edge in a saved graph carries no data of its own, and this is + // derived from the graph anyway. InteractEdge reads it to caption the edge + // rather than re-deriving both endpoints' node types per edge. + const edgesWithPeerFlag = useMemo( + () => edges.map((e) => (peerEdgeIds.has(e.id) ? { ...e, data: { ...e.data, isPeerEdge: true } } : e)), + [edges, peerEdgeIds], + ) + + // The model each agent will actually run on, resolved through its AI + // connector. Injected into the node's data rather than read here, because + // whether that model can be sent function schemas needs the provider's model + // list -- a query, which the node card subscribes to itself (see AgentNode's + // peerNeedsToolCalling). Only the *wiring* half belongs in this file. + const llmConfigByAgent = useMemo(() => { + const nodeById = new Map(nodes.map((n) => [n.id, n])) + const map = new Map() + for (const e of edges) { + if (e.targetHandle !== 'ai') continue + const config = (nodeById.get(e.source)?.data as LlmNodeData | undefined)?.config + if (config) map.set(e.target, { provider: config.provider, model: config.model }) + } + return map + }, [nodes, edges]) + + const agentNames = useMemo( + () => new Map(nodes.filter((n) => n.type === 'agent').map((n) => [n.id, (n.data as AgentNodeData).label || 'Agent'])), + [nodes], + ) + + // Every agent that could open a conversation -- it has at least one peer. + // The user picks which one they're addressing; the rest participate by being + // consulted. + const conversationCandidates = useMemo( + () => [...agentNames.keys()].filter((id) => (peerIdsByAgent.get(id)?.length ?? 0) > 0).map((id) => ({ id, name: agentNames.get(id)! })), + [agentNames, peerIdsByAgent], + ) + + useEffect(() => { + if (!conversationCandidates.some((a) => a.id === conversationEntryAgentId)) { + setConversationEntryAgentId(conversationCandidates[0]?.id ?? '') + } + }, [conversationCandidates, conversationEntryAgentId]) + const nodesWithRunStatus = useMemo((): Node[] => { return nodes.map((n) => { const patternHostId = patternHostIds.get(n.id) @@ -666,6 +761,8 @@ export const ProtocolCanvas = forwardRef 0, + llmConfig: n.type === 'agent' ? llmConfigByAgent.get(n.id) ?? null : null, // Only meaningful once the pattern is actually wired to an agent -- // an orphaned pattern node has no loop to warn about. hostHasNoTools: !!patternHostId && !agentIdsWithCallableTools.has(patternHostId), @@ -680,6 +777,8 @@ export const ProtocolCanvas = forwardRef )} + {/* Offered only once two agents are actually wired together -- + with nobody to consult, a conversation is just a slower + single-agent run. */} + {conversationCandidates.length > 0 && ( + + )}