From 98cc39dbefad091a33a023e55c86120fdde06c0c Mon Sep 17 00:00:00 2001 From: YuShifan <894402575bt@gmail.com> Date: Mon, 29 Jun 2026 17:44:08 +0800 Subject: [PATCH 1/3] feat: natural-language orchestration, DeepSeek provider, coordinator hardening Runtime API: - natural-language planner via the `skitter` agent: plan a new composed app from the live discovery registry, or route a request to an existing app so the normal coordinator session path handles execution - DeepSeek LLM provider (config, setup wizard, llm client, env/docs) - graph generation now requires every selected agent to be used Coordinator robustness: - guard the main MQTT message loop so one bad request can no longer tear down the coordinator - track agent online/offline status from discovery (a2a-status) and keep offline agents out of LLM planning - cap per-task event result/error payloads to bound MQTT packet size Shared helpers + tests: - llm.strip_code_fence and discovery.extract_app_tasks remove duplicated fence-stripping / app-extension parsing - tighten _resolve_app_ref to avoid resolving the wrong app - unit + e2e coverage for the above --- .env.example | 3 +- .gitignore | 6 + CONTRIBUTING.md | 2 +- docs/architecture.md | 4 + docs/usage.md | 4 +- skitter/config.py | 2 +- skitter/coordinator/registry.py | 18 +- skitter/coordinator/reply_handler.py | 17 +- skitter/coordinator/service.py | 148 ++++++++--- skitter/discovery.py | 38 +++ skitter/graph_gen.py | 33 ++- skitter/llm.py | 22 +- skitter/runtime_api.py | 264 +++++++++++++++++++- skitter/setup.py | 4 +- tests/test_e2e.py | 2 +- tests/unit/test_coordinator.py | 27 +- tests/unit/test_discovery.py | 18 ++ tests/unit/test_graph.py | 67 +++++ tests/unit/test_llm.py | 17 ++ tests/unit/test_runtime_api.py | 358 +++++++++++++++++++++++++++ tests/unit/test_setup.py | 13 + 21 files changed, 989 insertions(+), 78 deletions(-) diff --git a/.env.example b/.env.example index de305cb..e9d4bc1 100644 --- a/.env.example +++ b/.env.example @@ -18,12 +18,13 @@ MQTT_BROKER_URL=mqtt://localhost:1883 # API key for the configured provider. SKITTER_LLM_API_KEY= -# Which API to use: anthropic (default) or openai. +# Which API to use: anthropic (default), openai, openai-completions, or deepseek. # SKITTER_LLM_API=anthropic # Model name (no provider prefix needed). SKITTER_LLM_MODEL=claude-sonnet-4-6 # SKITTER_LLM_MODEL=gpt-5.4-mini +# SKITTER_LLM_MODEL=deepseek-chat # Optional: custom base URL for either API (e.g. proxy or self-hosted). # SKITTER_LLM_BASE_URL= diff --git a/.gitignore b/.gitignore index e6c056a..e6cc06d 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,12 @@ data/ tests/workspace/ tests/claude-state/ +# Frontend +node_modules/ +apps/*/node_modules/ +apps/*/dist/ +apps/*/.vite/ + # IDE .idea/ .vscode/ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 9918cb0..aff4509 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -140,7 +140,7 @@ llm: | `SKITTER_A2A_UNIT` | `default` | A2A topic unit segment | | `SKITTER_LLM_MODEL` | (empty) | Coordinator LLM model (overrides `llm.model` in config) | | `SKITTER_LLM_API_KEY` | (empty) | Coordinator LLM API key (overrides `llm.api_key` in config) | -| `SKITTER_LLM_API` | `anthropic` | Coordinator LLM provider: `anthropic`, `openai`, or `openai-completions` (overrides `llm.api` in config) | +| `SKITTER_LLM_API` | `anthropic` | Coordinator LLM provider: `anthropic`, `openai`, `openai-completions`, or `deepseek` (overrides `llm.api` in config) | | `SKITTER_LLM_BASE_URL` | (empty) | Custom endpoint URL for coordinator LLM (overrides `llm.base_url` in config) | | `CLAUDE_CODE_OAUTH_TOKEN` | (empty) | OAuth token for Claude Code agents (preferred; generate via `claude setup-token`). Stored per-agent in `~/.skitter/agents/.env` | | `ANTHROPIC_API_KEY` | (empty) | Anthropic API key for Claude Code agents (fallback). Stored per-agent in `~/.skitter/agents/.env` | diff --git a/docs/architecture.md b/docs/architecture.md index 7bf713c..805c130 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -190,6 +190,10 @@ The `skitter` agent handles structured queries: | `create app {json}` | Create composed app from agent IDs + instructions | | `delete app {id}` | Delete an app and all its versions, sessions, and tasks | +The same `skitter` agent also accepts natural language requests. It can plan a +new composed app from the current discovery registry, or route a request to an +existing app so the normal coordinator session path handles execution. + Session lifecycle events are published on `$a2a/v1/event/{org}/{unit}/skitter` for external consumers (e.g., dashboards). ## Database diff --git a/docs/usage.md b/docs/usage.md index 8a93990..a95c7df 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -79,7 +79,7 @@ Coordinator notes: - Standalone agents do not need coordinator LLM configuration - Composed apps need coordinator LLM configuration (model, API, key) for graph generation - The API key can be stored in `~/.skitter/config.yaml` (via `skitter setup`) or set as `SKITTER_LLM_API_KEY` env var -- Supported LLM APIs: `anthropic`, `openai` (Responses API), `openai-completions` (Chat Completions API, for 3rd-party OpenAI-compatible providers) +- Supported LLM APIs: `anthropic`, `openai` (Responses API), `openai-completions` (Chat Completions API, for 3rd-party OpenAI-compatible providers), `deepseek` (Chat Completions API, default model `deepseek-chat`) - Runtime auth for Claude Code or Codex is separate from coordinator LLM config ## Multi-turn Conversations @@ -108,7 +108,7 @@ Useful environment variables: - `SKITTER_HOME`: move config, agents, and the local database - `SKITTER_LLM_API_KEY`: coordinator LLM API key (overrides `llm.api_key` in config) - `SKITTER_LLM_MODEL`: coordinator model (overrides `llm.model` in config) -- `SKITTER_LLM_API`: coordinator provider: `anthropic`, `openai`, or `openai-completions` (overrides `llm.api` in config) +- `SKITTER_LLM_API`: coordinator provider: `anthropic`, `openai`, `openai-completions`, or `deepseek` (overrides `llm.api` in config) - `SKITTER_LLM_BASE_URL`: custom coordinator endpoint (overrides `llm.base_url` in config) - `MQTT_BROKER_URL`: override the broker URL diff --git a/skitter/config.py b/skitter/config.py index 145263f..0f62584 100644 --- a/skitter/config.py +++ b/skitter/config.py @@ -104,7 +104,7 @@ class DBConfig: @dataclass class LLMConfig: model: str = "" - api: str = "anthropic" # anthropic, openai, or openai-completions + api: str = "anthropic" # anthropic, openai, openai-completions, or deepseek base_url: str = "" api_key: str = "" diff --git a/skitter/coordinator/registry.py b/skitter/coordinator/registry.py index 25ba640..59165a6 100644 --- a/skitter/coordinator/registry.py +++ b/skitter/coordinator/registry.py @@ -12,21 +12,31 @@ class DiscoveryRegistry: def __init__(self) -> None: self._cards: dict[str, dict] = {} # agent_id -> card dict + self._statuses: dict[str, str] = {} # agent_id -> online/offline/unknown - def update(self, agent_id: str, card: dict) -> None: + def update(self, agent_id: str, card: dict, status: str = "unknown") -> None: self._cards[agent_id] = card + self._statuses[agent_id] = status log.info("Registry: updated card for %s", agent_id) def remove(self, agent_id: str) -> None: - if agent_id in self._cards: - del self._cards[agent_id] + removed = self._cards.pop(agent_id, None) + self._statuses.pop(agent_id, None) + if removed is not None: log.info("Registry: removed card for %s", agent_id) def get(self, agent_id: str) -> dict | None: return self._cards.get(agent_id) + def status(self, agent_id: str) -> str: + return self._statuses.get(agent_id, "unknown") + def list_agents(self) -> list[str]: - return [aid for aid, card in self._cards.items() if not is_app_card(card)] + return [ + aid + for aid, card in self._cards.items() + if not is_app_card(card) and self.status(aid) != "offline" + ] def list_apps(self) -> list[str]: return [aid for aid, card in self._cards.items() if is_app_card(card)] diff --git a/skitter/coordinator/reply_handler.py b/skitter/coordinator/reply_handler.py index edf962f..fc27bc6 100644 --- a/skitter/coordinator/reply_handler.py +++ b/skitter/coordinator/reply_handler.py @@ -133,7 +133,12 @@ async def complete_task( ) log.info("Task %s/%s completed", state.session_id, node_id) - await coord._publish_event("task_completed", state.session_id, task_id=node_id) + await coord._publish_event( + "task_completed", + state.session_id, + task_id=node_id, + data=coord._task_event_data(state, node_id, result=result), + ) # Check if session is complete if not state.inflight and not state.pending: @@ -186,7 +191,7 @@ async def fail_task( "task_failed", state.session_id, task_id=node_id, - data={"error": error[:200]}, + data=coord._task_event_data(state, node_id, error=error), ) # Check if session is done (all inflight finished) @@ -221,7 +226,11 @@ async def complete_session(coord: Coordinator, state: SessionState) -> None: artifact_text=result_text, ) - await coord._publish_event("session_completed", state.session_id) + await coord._publish_event( + "session_completed", + state.session_id, + data=coord._session_event_data(state), + ) coord._sessions.pop(state.session_id, None) coord._request_task_index.pop(state.request_task_id, None) coord._clear_context_active(state) @@ -250,7 +259,7 @@ async def fail_session(coord: Coordinator, state: SessionState, error: str) -> N await coord._publish_event( "session_failed", state.session_id, - data={"error": error[:200]}, + data={**coord._session_event_data(state), "error": error[:1000]}, ) coord._sessions.pop(state.session_id, None) coord._request_task_index.pop(state.request_task_id, None) diff --git a/skitter/coordinator/service.py b/skitter/coordinator/service.py index ccd0856..d4d0c14 100644 --- a/skitter/coordinator/service.py +++ b/skitter/coordinator/service.py @@ -34,6 +34,8 @@ CancelSessionResult, CreateAppResult, DeleteAppResult, + MessageResult, + RunAppResult, handle_query as runtime_query, coordinator_card, ) @@ -58,6 +60,7 @@ validate_a2a_request, ) from skitter.mqtt import ( + get_user_property, make_properties, mqtt_client_kwargs, ) @@ -144,6 +147,39 @@ async def _publish_event( "Failed to publish %s event for session %s", event_type, session_id ) + def _session_event_data(self, state: SessionState) -> dict: + return { + "request_task_id": state.request_task_id, + "app_id": state.app_id, + "context_id": state.context_id, + } + + def _task_event_data( + self, + state: SessionState, + node_id: str, + *, + result: str = "", + error: str = "", + ) -> dict: + task = state.graph.get(node_id) + data = { + **self._session_event_data(state), + "node_id": node_id, + } + if task: + data.update( + { + "agent": task.target.agent if task.target else task.agent, + "description": task.description, + } + ) + if result: + data["result"] = result[: self._MAX_RESULT_CHARS] + if error: + data["error"] = error[: self._MAX_RESULT_CHARS] + return data + # --- Conversation continuity --- async def _build_conversation_history(self, app_id: str, context_id: str) -> str: @@ -271,10 +307,10 @@ async def _dispatch_task(self, state: SessionState, node_id: str) -> None: context = build_context(state, task) if context: parts.append(context) - parts.append(task.description) user_request = state.variables.get("user_request", "") if user_request: - parts.append(f"User request: {user_request}") + parts.append(f"Workflow request for context only:\n{user_request}") + parts.append(f"Execute only this task:\n{task.description}") prompt = "\n\n".join(parts) correlation = uuid.uuid4().hex[:16] @@ -327,7 +363,12 @@ async def _dispatch_task(self, state: SessionState, node_id: str) -> None: target.agent, correlation, ) - await self._publish_event("task_started", state.session_id, task_id=node_id) + await self._publish_event( + "task_started", + state.session_id, + task_id=node_id, + data=self._task_event_data(state, node_id), + ) # --- Reply handling --- @@ -382,6 +423,18 @@ async def _handle_runtime_query( """ result = await runtime_query(self._adb, req.text, self._registry) + if isinstance(result, RunAppResult): + app_req = A2ARequest( + text=result.prompt, + request_id=req.request_id, + task_id=req.task_id, + context_id=req.context_id, + sender=req.sender or RUNTIME_AGENT_ID, + variables=req.variables, + ) + await self.handle_request(app_req, reply_topic, correlation, result.app_id) + return + try: if isinstance(result, CancelSessionResult): await self._cancel_session_cleanup(result.session_id) @@ -397,7 +450,7 @@ async def _handle_runtime_query( correlation, req.task_id, req.context_id or "", - artifact_text=json.dumps(result.to_dict()), + artifact_text=_runtime_artifact_text(result), ) async def _cancel_session_cleanup(self, session_id: str) -> None: @@ -557,7 +610,11 @@ async def _start_session(self, state: SessionState, label: str) -> None: await self._client.publish( state.caller_reply_topic, ack, qos=1, properties=props ) - await self._publish_event("session_created", state.session_id) + await self._publish_event( + "session_created", + state.session_id, + data=self._session_event_data(state), + ) await self.dispatch_ready(state) log.info( "%s session %s started (%d tasks)", @@ -814,8 +871,14 @@ async def _send_error( # --- Discovery subscription --- - def handle_discovery(self, topic: str, payload: bytes) -> None: - """Process a discovery card update from the broker.""" + def handle_discovery( + self, topic: str, payload: bytes, status: str = "unknown" + ) -> None: + """Process a discovery card update from the broker. + + ``status`` is the agent's ``a2a-status`` user property (online/offline); + offline agents are kept out of planning by the registry. + """ agent_id = topic.split("/")[-1] if agent_id == RUNTIME_AGENT_ID: return @@ -824,7 +887,7 @@ def handle_discovery(self, topic: str, payload: bytes) -> None: return try: card = parse_card(payload) - self._registry.update(agent_id, card) + self._registry.update(agent_id, card, status) except Exception: log.warning("Failed to parse discovery card for %s", agent_id) @@ -902,16 +965,6 @@ async def run(self) -> None: retain=True, ) - # Best-effort LLM connectivity check; warn but don't block - # (only create-app needs the LLM, not runtime queries or recovery) - try: - from skitter.llm import check as llm_check - - async with asyncio.timeout(10): - await llm_check() - except Exception as exc: - log.warning("LLM check failed (create-app will not work): %s", exc) - # Recover apps (subscribe + republish cards) and inflight sessions await self.recover() log.info("Coordinator ready (lock=%s)", instance_id) @@ -922,28 +975,37 @@ async def run(self) -> None: payload = payload_bytes.decode() if payload_bytes else "" log.debug("MQTT ← %s (%d bytes)", topic, len(payload)) - if "/discovery/" in topic: - self.handle_discovery(topic, payload_bytes or b"") - - elif "/request/" in topic and "/cancel" not in topic: - # Main connection only handles runtime API requests; - # app requests arrive on dedicated per-app connections. - validated = await validate_a2a_request( - mqtt_msg, client, log=log - ) - if not validated: - continue - req, caller_reply, caller_corr = validated - - await self._handle_runtime_query(req, caller_reply, caller_corr) - - elif "/reply/" in topic and "/skitter/" in topic: - if payload: - corr_bytes = getattr( - mqtt_msg.properties, "CorrelationData", None + try: + if "/discovery/" in topic: + status = get_user_property(mqtt_msg, "a2a-status") + self.handle_discovery( + topic, payload_bytes or b"", status or "unknown" ) - corr = corr_bytes.decode() if corr_bytes else "" - await self.handle_reply(topic, payload, corr) + + elif "/request/" in topic and "/cancel" not in topic: + # Main connection only handles runtime API requests; + # app requests arrive on dedicated per-app connections. + validated = await validate_a2a_request( + mqtt_msg, client, log=log + ) + if not validated: + continue + req, caller_reply, caller_corr = validated + + await self._handle_runtime_query( + req, caller_reply, caller_corr + ) + + elif "/reply/" in topic and "/skitter/" in topic: + if payload: + corr_bytes = getattr( + mqtt_msg.properties, "CorrelationData", None + ) + corr = corr_bytes.decode() if corr_bytes else "" + await self.handle_reply(topic, payload, corr) + except Exception: + # One bad message must never tear down the coordinator loop. + log.exception("Error handling MQTT message on %s", topic) finally: # Tear down per-app connections for app_id in list(self._app_tasks): @@ -976,6 +1038,14 @@ def _parse_agent_id_from_topic(topic: str) -> str: return parts[5] if len(parts) >= 6 else "" +def _runtime_artifact_text(result) -> str: + if isinstance(result, MessageResult): + return result.message + if isinstance(result, CreateAppResult) and result.message: + return result.message + return json.dumps(result.to_dict()) + + # --- Entry point --- diff --git a/skitter/discovery.py b/skitter/discovery.py index 47d46c0..4515dbf 100644 --- a/skitter/discovery.py +++ b/skitter/discovery.py @@ -5,6 +5,7 @@ from skitter.config import AgentDef, load_config as _load_config APP_EXTENSION_URI = "urn:skitter:app" +TASK_AGENT_EXTENSION_URI = "urn:skitter:task-agent" def build_card( @@ -91,3 +92,40 @@ def is_app_card(card: dict) -> bool: if ext.get("uri") == APP_EXTENSION_URI: return bool(ext.get("params", {}).get("tasks")) return False + + +def extract_app_tasks(card: dict) -> list[dict]: + """Return the normalized task list from an app card's app extension.""" + for ext in card.get("capabilities", {}).get("extensions", []): + if ext.get("uri") != APP_EXTENSION_URI: + continue + tasks = ext.get("params", {}).get("tasks", []) + if isinstance(tasks, list): + return [ + { + "id": str(task.get("id", "")), + "agent": str(task.get("agent", "")), + "description": str(task.get("description", "")), + "needs": task_needs(task), + "terminal": bool(task.get("terminal", False)), + } + for task in tasks + if isinstance(task, dict) + ] + return [] + + +def task_needs(task: dict) -> list[str]: + """Normalize a task's ``needs`` list to clean, non-empty string ids.""" + needs = task.get("needs", []) + if not isinstance(needs, list): + return [] + return [str(need).strip() for need in needs if str(need).strip()] + + +def is_task_agent_card(card: dict) -> bool: + """Detect task agents by presence of the Skitter task-agent extension.""" + for ext in card.get("capabilities", {}).get("extensions", []): + if ext.get("uri") == TASK_AGENT_EXTENSION_URI: + return True + return False diff --git a/skitter/graph_gen.py b/skitter/graph_gen.py index 2d74011..a0d26d7 100644 --- a/skitter/graph_gen.py +++ b/skitter/graph_gen.py @@ -3,7 +3,7 @@ import json import logging -from skitter.llm import complete +from skitter.llm import complete, strip_code_fence log = logging.getLogger("skitter.graph_gen") @@ -26,6 +26,8 @@ Rules: - Every agent reference must match an agent ID from the provided list. +- Use every provided agent exactly once unless the instructions explicitly say + not to use a provided agent. - The graph must be a DAG (no cycles). - Task IDs must be unique. - "needs" lists upstream dependencies (tasks whose results this task requires). @@ -34,7 +36,6 @@ - A terminal task must not be listed in any other task's "needs". - Every non-terminal task must be listed in at least one other task's "needs". - Omit "terminal" (or set it to false) for non-terminal tasks. -- Use each agent at most once unless the instructions explicitly require multiple uses. """ @@ -54,7 +55,12 @@ class GraphValidationError(Exception): """Raised when a generated graph fails validation.""" -def validate_graph(graph: dict, valid_agent_ids: set[str]) -> None: +def validate_graph( + graph: dict, + valid_agent_ids: set[str], + *, + required_agent_ids: set[str] | None = None, +) -> None: """Validate an orchestration graph. Raises GraphValidationError on failure.""" tasks = graph.get("tasks") if not tasks or not isinstance(tasks, list): @@ -83,6 +89,15 @@ def validate_graph(graph: dict, valid_agent_ids: set[str]) -> None: if need not in all_ids: raise GraphValidationError(f"Task '{tid}' needs unknown task '{need}'") + if required_agent_ids: + used_agent_ids = {str(t.get("agent", "")) for t in tasks} + missing_agent_ids = required_agent_ids - used_agent_ids + if missing_agent_ids: + raise GraphValidationError( + "Graph must use every selected agent; missing: " + f"{', '.join(sorted(missing_agent_ids))}" + ) + # Check for cycles using DFS on dependency edges adj: dict[str, list[str]] = {t.get("id"): t.get("needs", []) for t in tasks} visited: set[str] = set() @@ -129,25 +144,19 @@ async def generate_graph( agent_cards: dict[str, dict], *, model: str = "", + required_agent_ids: set[str] | None = None, ) -> dict: """Use LLM to generate an orchestration graph from instructions + agent cards. Validates the result and retries once on validation failure. """ valid_ids = set(agent_cards.keys()) - prompt = _build_prompt(instructions, agent_cards) for attempt in range(2): raw = await complete(prompt, system=_SYSTEM, model=model) - # Strip markdown fences if present - text = raw.strip() - if text.startswith("```"): - text = text.split("\n", 1)[1] if "\n" in text else text[3:] - if text.endswith("```"): - text = text[:-3] - text = text.strip() + text = strip_code_fence(raw) try: graph = json.loads(text) @@ -161,7 +170,7 @@ async def generate_graph( raise GraphValidationError(f"LLM returned invalid JSON: {e}") from e try: - validate_graph(graph, valid_ids) + validate_graph(graph, valid_ids, required_agent_ids=required_agent_ids) return graph except GraphValidationError as e: if attempt == 0: diff --git a/skitter/llm.py b/skitter/llm.py index 32889c3..cc196a9 100644 --- a/skitter/llm.py +++ b/skitter/llm.py @@ -10,10 +10,23 @@ log = logging.getLogger("skitter.llm") +DEEPSEEK_BASE_URL = "https://api.deepseek.com" + # env var name → cached SDK client _client_cache: dict[str, object] = {} +def strip_code_fence(text: str) -> str: + """Strip a leading/trailing markdown code fence from LLM output.""" + text = text.strip() + if text.startswith("```"): + text = text.split("\n", 1)[1] if "\n" in text else text[3:] + if text.endswith("```"): + text = text[:-3] + text = text.strip() + return text + + def _resolve_model(cfg: LLMConfig) -> str: if not cfg.model: raise ValueError( @@ -50,12 +63,13 @@ def _anthropic_client(cfg: LLMConfig): def _openai_client(cfg: LLMConfig): from openai import AsyncOpenAI - cache_key = f"openai:{cfg.base_url}" + base_url = cfg.base_url or (DEEPSEEK_BASE_URL if cfg.api == "deepseek" else "") + cache_key = f"{cfg.api}:{base_url}" if cache_key in _client_cache: return _client_cache[cache_key] kwargs: dict = {"api_key": _api_key(cfg)} - if cfg.base_url: - kwargs["base_url"] = cfg.base_url + if base_url: + kwargs["base_url"] = base_url client = AsyncOpenAI(**kwargs) _client_cache[cache_key] = client return client @@ -94,7 +108,7 @@ async def complete( try: if cfg.api == "openai": return await _complete_openai(prompt, system=system, model=model, cfg=cfg) - if cfg.api == "openai-completions": + if cfg.api in ("openai-completions", "deepseek"): return await _complete_openai_chat( prompt, system=system, model=model, cfg=cfg ) diff --git a/skitter/runtime_api.py b/skitter/runtime_api.py index 7fe743d..2b90245 100644 --- a/skitter/runtime_api.py +++ b/skitter/runtime_api.py @@ -18,6 +18,7 @@ import json import logging +import re from abc import ABC, abstractmethod from dataclasses import dataclass from typing import TYPE_CHECKING @@ -27,8 +28,14 @@ from skitter.a2a import TaskState from skitter.config import AgentDef from skitter.db import App, AppVersion, AsyncDB -from skitter.discovery import build_card +from skitter.discovery import ( + build_card, + extract_app_tasks, + is_task_agent_card, + task_needs, +) from skitter.graph_gen import GraphValidationError, generate_graph +from skitter.llm import complete, strip_code_fence if TYPE_CHECKING: from skitter.coordinator import DiscoveryRegistry @@ -68,6 +75,16 @@ def to_dict(self) -> dict: return {"error": self.message} +@dataclass +class MessageResult(QueryResult): + """Plain user-facing chat result.""" + + message: str + + def to_dict(self) -> dict: + return {"message": self.message} + + @dataclass class CancelSessionResult(QueryResult): """Session canceled; coordinator should clean up.""" @@ -85,6 +102,7 @@ class CreateAppResult(QueryResult): app_id: str version: int card_json: str + message: str = "" def to_dict(self) -> dict: return { @@ -106,12 +124,30 @@ def to_dict(self) -> dict: return {"deleted_app": self.app_id} +@dataclass +class RunAppResult(QueryResult): + """Run an existing app via the coordinator app execution path.""" + + app_id: str + prompt: str + reason: str = "" + + def to_dict(self) -> dict: + return { + "run_app": { + "app_id": self.app_id, + "prompt": self.prompt, + "reason": self.reason, + } + } + + def coordinator_card() -> dict: """Build the discovery card for the skitter agent.""" agent = AgentDef( id=AGENT_ID, name="Skitter", - description="Create and manage composed multi-agent apps", + description="Create, run, and coordinate composed multi-agent workflows", ) return build_card(agent) @@ -143,6 +179,9 @@ async def handle_query( if verb == "delete" and noun == "app" and arg: return await _delete_app(db, arg) + if registry: + return await _handle_natural_language(db, text.strip(), registry) + return ErrorResult(f"Unknown query: {text.strip()}") @@ -255,6 +294,223 @@ async def _delete_app(db: AsyncDB, app_id: str) -> QueryResult: return DeleteAppResult(app_id=app_id) +_NATURAL_LANGUAGE_SYSTEM = """\ +You are Skitter, an MQTT A2A workflow coordinator. + +The user talks to you in natural language. Decide whether to answer, create a +workflow app, or run an existing workflow app. + +Return only one JSON object with one of these shapes: + +{"action":"answer","message":"short user-facing answer"} + +{"action":"create_app","name":"workflow name","description":"short description","instructions":"natural language orchestration instructions","agents":["agent-id"]} + +{"action":"run_app","app_id":"existing-app-id","prompt":"what the user wants this app to do","reason":"short reason"} + +Rules: +- Use only agent IDs listed in available_agents. +- For create_app, choose all relevant agents. If the request is broad, include all available agents. +- Do not invent business-specific workflows. Use the user's words as the source of truth. +- If the user asks to enable, start, open, run, or execute an existing workflow, use run_app. +- If the user includes concrete runtime details, copy them verbatim into run_app.prompt. +- If no existing app clearly matches a run request, answer that the workflow needs to be created first. +- Keep answers concise and conversational. +""" + + +async def _handle_natural_language( + db: AsyncDB, text: str, registry: DiscoveryRegistry +) -> QueryResult: + """Turn natural language into runtime actions without UI-side rules.""" + app_summaries = await _list_app_summaries(db) + agent_summaries = _list_agent_summaries(registry) + + if not agent_summaries and not app_summaries: + return MessageResult( + "I do not see any available agents or workflows yet. Start agents first, " + "then ask me to create a workflow." + ) + + prompt = json.dumps( + { + "user_message": text, + "existing_apps": app_summaries, + "available_agents": agent_summaries, + }, + ensure_ascii=False, + indent=2, + ) + + try: + raw = await complete(prompt, system=_NATURAL_LANGUAGE_SYSTEM) + decision = _parse_llm_json(raw) + except Exception as exc: + log.exception("Runtime planner failed") + return MessageResult(f"I could not plan that request yet: {exc}") + + action = str(decision.get("action", "")).strip().lower() + if action == "answer": + message = str(decision.get("message", "")).strip() + return MessageResult( + message or "Tell me what workflow you want to create or run." + ) + + if action == "create_app": + return await _create_app_from_plan(db, decision, registry, agent_summaries) + + if action == "run_app": + return _run_app_from_plan(decision, app_summaries, fallback_prompt=text) + + return MessageResult( + "I can create workflows from your request, or run an existing workflow." + ) + + +async def _list_app_summaries(db: AsyncDB) -> list[dict]: + apps = await db.list_apps() + summaries = [] + for app in apps: + current = await db.get_current_version(app.id) + tasks = [] + if app.card_json: + try: + card = json.loads(app.card_json) + tasks = extract_app_tasks(card) + except Exception: + tasks = [] + summaries.append( + { + "id": app.id, + "name": app.name, + "description": app.description, + "current_version": current.version if current else None, + "tasks": tasks, + } + ) + return summaries + + +def _list_agent_summaries(registry: DiscoveryRegistry) -> list[dict]: + summaries = [] + for agent_id in registry.list_agents(): + if agent_id == AGENT_ID: + continue + card = registry.get(agent_id) or {} + if is_task_agent_card(card): + continue + summaries.append( + { + "id": agent_id, + "name": card.get("name", agent_id), + "description": card.get("description", ""), + "status": registry.status(agent_id), + "skills": [ + { + "id": skill.get("id", ""), + "name": skill.get("name", ""), + "description": skill.get("description", ""), + } + for skill in card.get("skills", []) + if isinstance(skill, dict) + ], + } + ) + return summaries + + +def _parse_llm_json(raw: str) -> dict: + data = json.loads(strip_code_fence(raw)) + if not isinstance(data, dict): + raise ValueError("planner returned a non-object JSON value") + return data + + +async def _create_app_from_plan( + db: AsyncDB, + decision: dict, + registry: DiscoveryRegistry, + agent_summaries: list[dict], +) -> QueryResult: + name = str(decision.get("name", "")).strip() + instructions = str(decision.get("instructions", "")).strip() + if not name or not instructions: + return MessageResult( + "Tell me the workflow name and what the agents should coordinate." + ) + + available_agent_ids = {str(agent["id"]) for agent in agent_summaries} + requested_agents = [ + str(agent_id) + for agent_id in decision.get("agents", []) + if str(agent_id) in available_agent_ids + ] + agent_ids = requested_agents or sorted(available_agent_ids) + if not agent_ids: + return MessageResult( + "I do not see any available agents to build that workflow." + ) + + spec = json.dumps( + { + "id": _stable_app_id(name), + "name": name, + "description": str(decision.get("description", "")).strip(), + "instructions": instructions, + "agents": agent_ids, + }, + ensure_ascii=False, + ) + result = await _handle_create_app(db, spec, registry) + if isinstance(result, CreateAppResult): + result.message = ( + f"Created workflow '{name}' with {len(agent_ids)} agent" + f"{'' if len(agent_ids) == 1 else 's'}. Ask me to run it when you are ready." + ) + return result + + +def _run_app_from_plan( + decision: dict, app_summaries: list[dict], *, fallback_prompt: str +) -> QueryResult: + app_ref = str(decision.get("app_id", "")).strip() + app = _resolve_app_ref(app_ref, app_summaries) + if not app: + return MessageResult( + "I do not see a matching workflow yet. Ask me to create it first." + ) + + prompt = str(decision.get("prompt", "")).strip() or fallback_prompt + return RunAppResult( + app_id=str(app["id"]), + prompt=prompt, + reason=str(decision.get("reason", "")).strip(), + ) + + +def _resolve_app_ref(ref: str, app_summaries: list[dict]) -> dict | None: + normalized = ref.casefold().strip() + if not normalized: + return None + for app in app_summaries: + if str(app["id"]).casefold() == normalized: + return app + for app in app_summaries: + if str(app["name"]).casefold() == normalized: + return app + # Fuzzy fallback: only an unambiguous substring match on the name. Never + # match the free-text description, and never guess when several names match. + name_matches = [ + app for app in app_summaries if normalized in str(app["name"]).casefold() + ] + return name_matches[0] if len(name_matches) == 1 else None + + +def _stable_app_id(name: str) -> str: + slug = re.sub(r"[^a-z0-9]+", "-", name.strip().lower()).strip("-") + return slug[:48] or uuid.uuid4().hex[:12] + + async def create_app( db: AsyncDB, *, @@ -300,6 +556,8 @@ async def create_app( "id": t["id"], "agent": t.get("agent", ""), "description": t.get("description", ""), + "needs": task_needs(t), + "terminal": bool(t.get("terminal", False)), } for t in tasks ], @@ -358,7 +616,7 @@ async def _handle_create_app( # Generate orchestration graph via LLM try: - graph = await generate_graph(instructions, cards) + graph = await generate_graph(instructions, cards, required_agent_ids=set(cards)) except GraphValidationError as e: return ErrorResult(f"Graph generation failed: {e}") except Exception as e: diff --git a/skitter/setup.py b/skitter/setup.py index 6666d7b..4e723eb 100644 --- a/skitter/setup.py +++ b/skitter/setup.py @@ -95,6 +95,7 @@ def _collect_llm( "anthropic": "claude-sonnet-4-6", "openai": "gpt-5.4-mini", "openai-completions": "gpt-5.4-mini", + "deepseek": "deepseek-chat", } if non_interactive: @@ -115,9 +116,10 @@ def _collect_llm( print("\nWhich LLM API?") print(" (openai-completions: for 3rd-party OpenAI-compatible providers)") + print(" (deepseek: DeepSeek chat completions)") api = _prompt_choice( "API", - ["anthropic", "openai", "openai-completions"], + ["anthropic", "openai", "openai-completions", "deepseek"], default=existing.get("api", "anthropic"), ) diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 7d6a542..ff266c1 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -135,7 +135,7 @@ def mock_graph(): """Fixture to set the graph that generate_graph will return.""" container: dict = {} - async def _mock(instructions, agent_cards, *, model=""): + async def _mock(instructions, agent_cards, *, model="", required_agent_ids=None): return dict(container["graph"]) with patch("skitter.runtime_api.generate_graph", new=_mock): diff --git a/tests/unit/test_coordinator.py b/tests/unit/test_coordinator.py index e960b2b..372c229 100644 --- a/tests/unit/test_coordinator.py +++ b/tests/unit/test_coordinator.py @@ -698,8 +698,8 @@ async def test_forward_stream_qos_1(self): ) @pytest.mark.asyncio - async def test_dispatch_includes_user_request_in_prompt(self): - """Dispatched A2A request must append user request to prompt text.""" + async def test_dispatch_keeps_task_instruction_last(self): + """User request is context; the task instruction remains authoritative.""" sup, mock_client = await self._make_coordinator_with_app() req = A2ARequest(text="find latest news", request_id="r1") @@ -720,11 +720,13 @@ async def test_dispatch_includes_user_request_in_prompt(self): ] assert len(dispatched_payloads) == 1 prompt_text = dispatched_payloads[0]["params"]["message"]["parts"][0]["text"] - assert "User request: find latest news" in prompt_text + assert "Workflow request for context only" in prompt_text + assert "find latest news" in prompt_text + assert prompt_text.endswith("Execute only this task:\nDo it") @pytest.mark.asyncio async def test_dispatch_omits_user_request_when_empty(self): - """Dispatched prompt must not contain 'User request:' when request text is empty.""" + """Dispatched prompt must omit workflow context when request text is empty.""" sup, mock_client = await self._make_coordinator_with_app() req = A2ARequest(text="", request_id="r1") @@ -745,7 +747,8 @@ async def test_dispatch_omits_user_request_when_empty(self): ] assert len(dispatched_payloads) == 1 prompt_text = dispatched_payloads[0]["params"]["message"]["parts"][0]["text"] - assert "User request:" not in prompt_text + assert "Workflow request for context only" not in prompt_text + assert prompt_text.endswith("Execute only this task:\nDo it") @pytest.mark.asyncio async def test_dispatched_request_has_context_id(self): @@ -944,6 +947,20 @@ def test_list_agents_vs_apps(self): assert "app1" in reg.list_apps() assert "agent1" not in reg.list_apps() + def test_offline_agents_excluded_from_list_agents(self): + from skitter.coordinator import DiscoveryRegistry + + reg = DiscoveryRegistry() + reg.update("online-agent", {"name": "Online"}, "online") + reg.update("quiet-agent", {"name": "Quiet"}) # defaults to "unknown" + reg.update("offline-agent", {"name": "Offline"}, "offline") + + agents = reg.list_agents() + assert "online-agent" in agents + assert "quiet-agent" in agents # unknown is still selectable + assert "offline-agent" not in agents + assert reg.status("offline-agent") == "offline" + # --- Helpers for write-ahead, recovery, and dedup edge case tests --- diff --git a/tests/unit/test_discovery.py b/tests/unit/test_discovery.py index 83d5ecc..03bef4d 100644 --- a/tests/unit/test_discovery.py +++ b/tests/unit/test_discovery.py @@ -130,6 +130,24 @@ def test_is_app_card(self): } ) + def test_is_task_agent_card(self): + from skitter.discovery import is_task_agent_card + + assert not is_task_agent_card({"name": "Agent"}) + assert not is_task_agent_card({"capabilities": {}}) + assert is_task_agent_card( + { + "capabilities": { + "extensions": [ + { + "uri": "urn:skitter:task-agent", + "params": {"kind": "mock-task-agent"}, + } + ] + } + } + ) + class TestDiscoveryWildcard: def test_default_org_unit(self): diff --git a/tests/unit/test_graph.py b/tests/unit/test_graph.py index c2e53cd..62fee7d 100644 --- a/tests/unit/test_graph.py +++ b/tests/unit/test_graph.py @@ -106,6 +106,21 @@ def test_terminal_has_dependents(self): with pytest.raises(GraphValidationError, match="must not have dependents"): validate_graph(graph, {"a", "b"}) + def test_required_agent_ids_must_be_used(self): + from skitter.graph_gen import GraphValidationError, validate_graph + + graph = { + "tasks": [ + {"id": "read", "agent": "reader", "needs": [], "terminal": True}, + ] + } + with pytest.raises(GraphValidationError, match="missing: analyzer"): + validate_graph( + graph, + {"reader", "analyzer"}, + required_agent_ids={"reader", "analyzer"}, + ) + class TestGraphGeneration: def _make_cards(self): @@ -209,6 +224,58 @@ async def test_generate_retries_on_validation_error(self): assert mock_llm.call_count == 2 assert graph["tasks"][0]["agent"] == "agent-reader-001" + @pytest.mark.asyncio + async def test_generate_retries_when_required_agent_missing(self): + from skitter.graph_gen import generate_graph + + missing_analyzer = json.dumps( + { + "tasks": [ + { + "id": "read", + "agent": "agent-reader-001", + "description": "Read", + "needs": [], + "terminal": True, + } + ] + } + ) + complete_graph = json.dumps( + { + "tasks": [ + { + "id": "read", + "agent": "agent-reader-001", + "description": "Read", + "needs": [], + }, + { + "id": "analyze", + "agent": "agent-analyzer-002", + "description": "Analyze", + "needs": ["read"], + "terminal": True, + }, + ] + } + ) + + with patch("skitter.graph_gen.complete", new_callable=AsyncMock) as mock_llm: + mock_llm.side_effect = [missing_analyzer, complete_graph] + graph = await generate_graph( + "Read and analyze", + self._make_cards(), + model="test", + required_agent_ids={"agent-reader-001", "agent-analyzer-002"}, + ) + + assert mock_llm.call_count == 2 + assert {task["agent"] for task in graph["tasks"]} == { + "agent-reader-001", + "agent-analyzer-002", + } + @pytest.mark.asyncio async def test_generate_fails_after_retries(self): diff --git a/tests/unit/test_llm.py b/tests/unit/test_llm.py index f06ad3e..6e6426a 100644 --- a/tests/unit/test_llm.py +++ b/tests/unit/test_llm.py @@ -195,6 +195,23 @@ async def test_complete_openai_completions_with_system(self): assert messages[0] == {"role": "system", "content": "be helpful"} assert messages[1] == {"role": "user", "content": "hello"} + @pytest.mark.asyncio + async def test_complete_deepseek_uses_chat_api_and_default_base_url(self): + from skitter.config import LLMConfig + from skitter.llm import DEEPSEEK_BASE_URL, complete + + mock_client, mock_create = self._mock_openai_chat("deepseek response") + cfg = LLMConfig(model="deepseek-chat", api="deepseek", api_key="test-key") + with ( + patch("openai.AsyncOpenAI", return_value=mock_client) as mock_cls, + patch("skitter.llm.load_config", return_value=MagicMock(llm=cfg)), + ): + result = await complete("hello", model="deepseek-chat") + + assert result == "deepseek response" + assert mock_cls.call_args.kwargs["base_url"] == DEEPSEEK_BASE_URL + mock_create.assert_called_once() + @pytest.mark.asyncio async def test_complete_none_content_raises(self): from skitter.config import LLMConfig diff --git a/tests/unit/test_runtime_api.py b/tests/unit/test_runtime_api.py index fd27c78..39325b8 100644 --- a/tests/unit/test_runtime_api.py +++ b/tests/unit/test_runtime_api.py @@ -52,6 +52,8 @@ async def test_create_app(self): if e["uri"] == "urn:skitter:app" ) assert len(wf["params"]["tasks"]) == 1 + assert wf["params"]["tasks"][0]["needs"] == [] + assert wf["params"]["tasks"][0]["terminal"] is True @pytest.mark.asyncio async def test_provided_app_id(self): @@ -359,6 +361,28 @@ async def test_create_app(self): assert isinstance(result, CreateAppResult) assert result.version == 1 assert result.card_json + card = json.loads(result.card_json) + wf = next( + e + for e in card["capabilities"]["extensions"] + if e["uri"] == "urn:skitter:app" + ) + assert wf["params"]["tasks"] == [ + { + "id": "read", + "agent": "reader", + "description": "Read", + "needs": [], + "terminal": False, + }, + { + "id": "analyze", + "agent": "analyzer", + "description": "Analyze", + "needs": ["read"], + "terminal": True, + }, + ] # Verify DB state app = self.db.get_app(result.app_id) @@ -396,6 +420,221 @@ async def test_create_app_no_registry(self): assert isinstance(result, ErrorResult) assert "registry" in result.message.lower() + @pytest.mark.asyncio + async def test_natural_language_creates_app(self): + from skitter.coordinator import DiscoveryRegistry + from skitter.runtime_api import CreateAppResult, handle_query + + registry = DiscoveryRegistry() + registry.update( + "lock", + {"name": "Lock", "description": "Controls a door lock"}, + ) + registry.update( + "fan", + {"name": "Fan", "description": "Controls a smart fan"}, + ) + + graph = { + "tasks": [ + { + "id": "lock", + "agent": "lock", + "description": "Lock the door", + "needs": [], + }, + { + "id": "fan", + "agent": "fan", + "description": "Turn off the fan", + "needs": ["lock"], + "terminal": True, + }, + ] + } + + decision = { + "action": "create_app", + "name": "Leave home mode", + "description": "Coordinate devices when leaving home", + "instructions": "Lock the door, then turn off the fan.", + "agents": ["lock", "fan"], + } + with ( + patch( + "skitter.runtime_api.complete", new_callable=AsyncMock + ) as mock_complete, + patch( + "skitter.runtime_api.generate_graph", new_callable=AsyncMock + ) as mock_gen, + ): + mock_complete.return_value = json.dumps(decision) + mock_gen.return_value = graph + result = await handle_query(self.adb, "Create leave home mode", registry) + + assert isinstance(result, CreateAppResult) + assert result.message.startswith("Created workflow") + app = self.db.get_app(result.app_id) + assert app is not None + assert app.name == "Leave home mode" + + @pytest.mark.asyncio + async def test_natural_language_create_excludes_task_agents(self): + from skitter.coordinator import DiscoveryRegistry + from skitter.runtime_api import CreateAppResult, handle_query + + registry = DiscoveryRegistry() + registry.update("lock", {"name": "Lock", "description": "Controls a door lock"}) + registry.update("fan", {"name": "Fan", "description": "Controls a smart fan"}) + registry.update( + "smart-home-task", + { + "name": "Smart Home Task Agent", + "description": "Coordinates smart home scenes", + "capabilities": { + "extensions": [ + { + "uri": "urn:skitter:task-agent", + "params": {"kind": "mock-task-agent"}, + } + ] + }, + }, + ) + + decision = { + "action": "create_app", + "name": "Home mode", + "description": "Coordinate home devices", + "instructions": "Unlock the door, then turn on the fan.", + "agents": ["lock", "fan"], + } + graph = { + "tasks": [ + { + "id": "unlock", + "agent": "lock", + "description": "Unlock the door", + "needs": [], + }, + { + "id": "fan", + "agent": "fan", + "description": "Turn on the fan", + "needs": ["unlock"], + "terminal": True, + }, + ] + } + + with ( + patch( + "skitter.runtime_api.complete", new_callable=AsyncMock + ) as mock_complete, + patch( + "skitter.runtime_api.generate_graph", new_callable=AsyncMock + ) as mock_gen, + ): + mock_complete.return_value = json.dumps(decision) + mock_gen.return_value = graph + result = await handle_query(self.adb, "Create home mode", registry) + + assert isinstance(result, CreateAppResult) + planner_payload = json.loads(mock_complete.call_args.args[0]) + assert {agent["id"] for agent in planner_payload["available_agents"]} == { + "lock", + "fan", + } + assert "smart-home-task" not in mock_gen.call_args.args[1] + + @pytest.mark.asyncio + async def test_natural_language_runs_existing_app(self): + from skitter.coordinator import DiscoveryRegistry + from skitter.runtime_api import RunAppResult, create_app, handle_query + + await create_app( + self.adb, + app_id="leave-home-mode", + name="Leave home mode", + description="Coordinate devices when leaving home", + graph={ + "tasks": [ + { + "id": "step", + "agent": "lock", + "description": "Do it", + "needs": [], + "terminal": True, + } + ] + }, + ) + registry = DiscoveryRegistry() + decision = { + "action": "run_app", + "app_id": "leave-home-mode", + "prompt": "Turn on leave home mode now.", + "reason": "The user asked to start the workflow.", + } + with patch( + "skitter.runtime_api.complete", new_callable=AsyncMock + ) as mock_complete: + mock_complete.return_value = json.dumps(decision) + result = await handle_query(self.adb, "开启离家模式", registry) + + assert isinstance(result, RunAppResult) + assert result.app_id == "leave-home-mode" + assert result.prompt == "Turn on leave home mode now." + + @pytest.mark.asyncio + async def test_natural_language_run_preserves_concrete_runtime_details(self): + from skitter.coordinator import DiscoveryRegistry + from skitter.runtime_api import RunAppResult, create_app, handle_query + + await create_app( + self.adb, + app_id="leave-home-mode", + name="Leave home mode", + description="Coordinate devices when leaving home", + graph={ + "tasks": [ + { + "id": "step", + "agent": "lock", + "description": "Do it", + "needs": [], + "terminal": True, + } + ] + }, + ) + registry = DiscoveryRegistry() + prompt = """开启离家模式 + +Confirmed device bindings for workflow "Leave home mode": +- Lock the door via Lock: device_id=lock-device-1, device_name=Front Door + +Use exactly these device_id values for the matching workflow steps. Do not choose other devices.""" + decision = { + "action": "run_app", + "app_id": "leave-home-mode", + "prompt": prompt, + "reason": "The user confirmed concrete runtime details.", + } + with patch( + "skitter.runtime_api.complete", new_callable=AsyncMock + ) as mock_complete: + mock_complete.return_value = json.dumps(decision) + result = await handle_query(self.adb, prompt, registry) + + assert isinstance(result, RunAppResult) + assert result.app_id == "leave-home-mode" + assert result.prompt == prompt + assert ( + "concrete runtime details" + in mock_complete.call_args.kwargs["system"].lower() + ) + class TestCoordinatorRuntimeRouting: """Test that the coordinator routes runtime queries correctly.""" @@ -420,6 +659,20 @@ def test_handle_discovery_skips_runtime(self): ) assert sup.registry.get(AGENT_ID) is None + def test_handle_discovery_excludes_offline_agent(self): + from skitter.coordinator import Coordinator + + sup = Coordinator(self.db) + topic = "$a2a/v1/discovery/skitter/default/lock-agent" + card = b'{"name":"Lock"}' + sup.handle_discovery(topic, card, "online") + assert "lock-agent" in sup.registry.list_agents() + # Agent republishes its (non-empty) card with offline status via LWT or + # graceful shutdown; it must drop out of the planner's selectable set. + sup.handle_discovery(topic, card, "offline") + assert "lock-agent" not in sup.registry.list_agents() + assert sup.registry.status("lock-agent") == "offline" + @pytest.mark.asyncio async def test_publish_event_structure(self): """Verify _publish_event builds correct payload.""" @@ -442,6 +695,58 @@ async def test_publish_event_structure(self): assert payload["task_id"] == "research" assert "timestamp" in payload + @pytest.mark.asyncio + async def test_runtime_run_app_result_routes_to_app(self): + """A natural-language run decision should enter the app execution path.""" + from skitter.coordinator import Coordinator + from skitter.runtime_api import RunAppResult, create_app + from skitter.db import AsyncDB + + sup = Coordinator(self.db) + mock_client = MagicMock() + mock_client.publish = AsyncMock() + mock_client.subscribe = AsyncMock() + sup._client = mock_client + + await create_app( + AsyncDB(self.db), + app_id="test-app", + name="Test App", + graph={ + "tasks": [ + { + "id": "step", + "agent": "researcher", + "description": "Do it", + "needs": [], + "terminal": True, + } + ] + }, + ) + + req = A2ARequest( + text="run the app", + request_id="q1", + task_id="request-task-1", + context_id="ctx-1", + ) + with patch( + "skitter.coordinator.service.runtime_query", new_callable=AsyncMock + ) as mock_runtime_query: + mock_runtime_query.return_value = RunAppResult( + app_id="test-app", prompt="run the app" + ) + await sup._handle_runtime_query(req, "reply/q", "corr-q") + + assert self.db.get_session_by_request_task_id("request-task-1") is not None + dispatch_calls = [ + call + for call in mock_client.publish.call_args_list + if "/request/" in str(call.args[0]) and "researcher" in str(call.args[0]) + ] + assert len(dispatch_calls) == 1 + @pytest.mark.asyncio async def test_publish_event_no_client(self): """No crash when client is None.""" @@ -535,6 +840,9 @@ async def test_session_lifecycle_events(self): assert "task_started" in event_types # session_created must come before task_started assert event_types.index("session_created") < event_types.index("task_started") + started_event = next(e for e in event_calls if e["event"] == "task_started") + assert started_event["data"]["request_task_id"] == req.task_id + assert started_event["data"]["agent"] == "researcher" # Simulate research task completion mock_client.publish.reset_mock() @@ -902,3 +1210,53 @@ async def test_delete_nonexistent_app(self): artifact = au["artifact"]["parts"][0]["text"] result = json.loads(artifact) assert "not found" in result["error"].lower() + + +class TestResolveAppRef: + def _apps(self): + return [ + { + "id": "leave-home", + "name": "Leave home mode", + "description": "lock doors", + }, + {"id": "mtg-notes", "name": "Meeting notes", "description": "summarize"}, + ] + + def test_exact_id_match(self): + from skitter.runtime_api import _resolve_app_ref + + app = _resolve_app_ref("mtg-notes", self._apps()) + assert app and app["id"] == "mtg-notes" + + def test_exact_name_match_is_case_insensitive(self): + from skitter.runtime_api import _resolve_app_ref + + app = _resolve_app_ref("leave home mode", self._apps()) + assert app and app["id"] == "leave-home" + + def test_unique_name_substring_matches(self): + from skitter.runtime_api import _resolve_app_ref + + app = _resolve_app_ref("meeting", self._apps()) + assert app and app["id"] == "mtg-notes" + + def test_ambiguous_substring_returns_none(self): + from skitter.runtime_api import _resolve_app_ref + + apps = [ + {"id": "a1", "name": "Morning notes", "description": ""}, + {"id": "a2", "name": "Evening notes", "description": ""}, + ] + assert _resolve_app_ref("notes", apps) is None + + def test_description_only_match_is_rejected(self): + from skitter.runtime_api import _resolve_app_ref + + # "summarize" appears only in a description, never in a name. + assert _resolve_app_ref("summarize", self._apps()) is None + + def test_blank_ref_returns_none(self): + from skitter.runtime_api import _resolve_app_ref + + assert _resolve_app_ref(" ", self._apps()) is None diff --git a/tests/unit/test_setup.py b/tests/unit/test_setup.py index 2987bfd..a5fff09 100644 --- a/tests/unit/test_setup.py +++ b/tests/unit/test_setup.py @@ -43,6 +43,19 @@ def test_respects_custom_api(self): assert result is not None assert result["api"] == "openai" + def test_deepseek_defaults_to_deepseek_chat(self): + from skitter.setup import _collect_llm + + env = { + "SKITTER_LLM_API": "deepseek", + "SKITTER_LLM_API_KEY": "sk-deepseek", + } + with patch.dict("os.environ", env, clear=True): + result = _collect_llm(non_interactive=True, standalone=False, existing={}) + assert result is not None + assert result["api"] == "deepseek" + assert result["model"] == "deepseek-chat" + class TestSetupVerifyPassesLLMConfig: """_verify must pass LLM config to check() without mutating os.environ.""" From 49d9668f7d7f6fea0063bf2549b4ea0ce50e50c4 Mon Sep 17 00:00:00 2001 From: YuShifan <894402575bt@gmail.com> Date: Mon, 29 Jun 2026 23:11:20 +0800 Subject: [PATCH 2/3] refactor: drop dedicated DeepSeek provider; harden planner + graph validation - remove the DeepSeek LLM provider special-casing (base_url default, routing, setup choice, tests). DeepSeek is OpenAI-compatible, so it works via `openai-completions` + base_url, now documented as such (review feedback) - runtime planner: stop echoing the raw exception in the user-facing message (it is already logged) - validate_graph: reject a non-list `needs` with a clear error instead of iterating it character by character; add a regression test --- .env.example | 3 +-- CONTRIBUTING.md | 2 +- docs/usage.md | 4 ++-- skitter/config.py | 2 +- skitter/graph_gen.py | 5 ++++- skitter/llm.py | 6 ++---- skitter/runtime_api.py | 6 ++++-- skitter/setup.py | 4 +--- tests/unit/test_graph.py | 13 +++++++++++++ tests/unit/test_llm.py | 17 ----------------- tests/unit/test_setup.py | 13 ------------- 11 files changed, 29 insertions(+), 46 deletions(-) diff --git a/.env.example b/.env.example index e9d4bc1..7dcf41e 100644 --- a/.env.example +++ b/.env.example @@ -18,13 +18,12 @@ MQTT_BROKER_URL=mqtt://localhost:1883 # API key for the configured provider. SKITTER_LLM_API_KEY= -# Which API to use: anthropic (default), openai, openai-completions, or deepseek. +# Which API to use: anthropic (default), openai, or openai-completions. # SKITTER_LLM_API=anthropic # Model name (no provider prefix needed). SKITTER_LLM_MODEL=claude-sonnet-4-6 # SKITTER_LLM_MODEL=gpt-5.4-mini -# SKITTER_LLM_MODEL=deepseek-chat # Optional: custom base URL for either API (e.g. proxy or self-hosted). # SKITTER_LLM_BASE_URL= diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index aff4509..9918cb0 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -140,7 +140,7 @@ llm: | `SKITTER_A2A_UNIT` | `default` | A2A topic unit segment | | `SKITTER_LLM_MODEL` | (empty) | Coordinator LLM model (overrides `llm.model` in config) | | `SKITTER_LLM_API_KEY` | (empty) | Coordinator LLM API key (overrides `llm.api_key` in config) | -| `SKITTER_LLM_API` | `anthropic` | Coordinator LLM provider: `anthropic`, `openai`, `openai-completions`, or `deepseek` (overrides `llm.api` in config) | +| `SKITTER_LLM_API` | `anthropic` | Coordinator LLM provider: `anthropic`, `openai`, or `openai-completions` (overrides `llm.api` in config) | | `SKITTER_LLM_BASE_URL` | (empty) | Custom endpoint URL for coordinator LLM (overrides `llm.base_url` in config) | | `CLAUDE_CODE_OAUTH_TOKEN` | (empty) | OAuth token for Claude Code agents (preferred; generate via `claude setup-token`). Stored per-agent in `~/.skitter/agents/.env` | | `ANTHROPIC_API_KEY` | (empty) | Anthropic API key for Claude Code agents (fallback). Stored per-agent in `~/.skitter/agents/.env` | diff --git a/docs/usage.md b/docs/usage.md index a95c7df..142dbe4 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -79,7 +79,7 @@ Coordinator notes: - Standalone agents do not need coordinator LLM configuration - Composed apps need coordinator LLM configuration (model, API, key) for graph generation - The API key can be stored in `~/.skitter/config.yaml` (via `skitter setup`) or set as `SKITTER_LLM_API_KEY` env var -- Supported LLM APIs: `anthropic`, `openai` (Responses API), `openai-completions` (Chat Completions API, for 3rd-party OpenAI-compatible providers), `deepseek` (Chat Completions API, default model `deepseek-chat`) +- Supported LLM APIs: `anthropic`, `openai` (Responses API), `openai-completions` (Chat Completions API, for 3rd-party OpenAI-compatible providers such as DeepSeek: set `base_url`, e.g. `https://api.deepseek.com`) - Runtime auth for Claude Code or Codex is separate from coordinator LLM config ## Multi-turn Conversations @@ -108,7 +108,7 @@ Useful environment variables: - `SKITTER_HOME`: move config, agents, and the local database - `SKITTER_LLM_API_KEY`: coordinator LLM API key (overrides `llm.api_key` in config) - `SKITTER_LLM_MODEL`: coordinator model (overrides `llm.model` in config) -- `SKITTER_LLM_API`: coordinator provider: `anthropic`, `openai`, `openai-completions`, or `deepseek` (overrides `llm.api` in config) +- `SKITTER_LLM_API`: coordinator provider: `anthropic`, `openai`, or `openai-completions` (overrides `llm.api` in config) - `SKITTER_LLM_BASE_URL`: custom coordinator endpoint (overrides `llm.base_url` in config) - `MQTT_BROKER_URL`: override the broker URL diff --git a/skitter/config.py b/skitter/config.py index 0f62584..145263f 100644 --- a/skitter/config.py +++ b/skitter/config.py @@ -104,7 +104,7 @@ class DBConfig: @dataclass class LLMConfig: model: str = "" - api: str = "anthropic" # anthropic, openai, openai-completions, or deepseek + api: str = "anthropic" # anthropic, openai, or openai-completions base_url: str = "" api_key: str = "" diff --git a/skitter/graph_gen.py b/skitter/graph_gen.py index a0d26d7..f5d3edb 100644 --- a/skitter/graph_gen.py +++ b/skitter/graph_gen.py @@ -85,7 +85,10 @@ def validate_graph( f"Task '{tid}' references unknown agent '{agent}'" ) - for need in t.get("needs", []): + needs = t.get("needs", []) + if not isinstance(needs, list): + raise GraphValidationError(f"Task '{tid}' has a non-list 'needs' field") + for need in needs: if need not in all_ids: raise GraphValidationError(f"Task '{tid}' needs unknown task '{need}'") diff --git a/skitter/llm.py b/skitter/llm.py index cc196a9..b177d71 100644 --- a/skitter/llm.py +++ b/skitter/llm.py @@ -10,8 +10,6 @@ log = logging.getLogger("skitter.llm") -DEEPSEEK_BASE_URL = "https://api.deepseek.com" - # env var name → cached SDK client _client_cache: dict[str, object] = {} @@ -63,7 +61,7 @@ def _anthropic_client(cfg: LLMConfig): def _openai_client(cfg: LLMConfig): from openai import AsyncOpenAI - base_url = cfg.base_url or (DEEPSEEK_BASE_URL if cfg.api == "deepseek" else "") + base_url = cfg.base_url cache_key = f"{cfg.api}:{base_url}" if cache_key in _client_cache: return _client_cache[cache_key] @@ -108,7 +106,7 @@ async def complete( try: if cfg.api == "openai": return await _complete_openai(prompt, system=system, model=model, cfg=cfg) - if cfg.api in ("openai-completions", "deepseek"): + if cfg.api == "openai-completions": return await _complete_openai_chat( prompt, system=system, model=model, cfg=cfg ) diff --git a/skitter/runtime_api.py b/skitter/runtime_api.py index 2b90245..8ba6ba3 100644 --- a/skitter/runtime_api.py +++ b/skitter/runtime_api.py @@ -345,9 +345,11 @@ async def _handle_natural_language( try: raw = await complete(prompt, system=_NATURAL_LANGUAGE_SYSTEM) decision = _parse_llm_json(raw) - except Exception as exc: + except Exception: log.exception("Runtime planner failed") - return MessageResult(f"I could not plan that request yet: {exc}") + return MessageResult( + "I could not plan that request yet. Please rephrase and try again." + ) action = str(decision.get("action", "")).strip().lower() if action == "answer": diff --git a/skitter/setup.py b/skitter/setup.py index 4e723eb..6666d7b 100644 --- a/skitter/setup.py +++ b/skitter/setup.py @@ -95,7 +95,6 @@ def _collect_llm( "anthropic": "claude-sonnet-4-6", "openai": "gpt-5.4-mini", "openai-completions": "gpt-5.4-mini", - "deepseek": "deepseek-chat", } if non_interactive: @@ -116,10 +115,9 @@ def _collect_llm( print("\nWhich LLM API?") print(" (openai-completions: for 3rd-party OpenAI-compatible providers)") - print(" (deepseek: DeepSeek chat completions)") api = _prompt_choice( "API", - ["anthropic", "openai", "openai-completions", "deepseek"], + ["anthropic", "openai", "openai-completions"], default=existing.get("api", "anthropic"), ) diff --git a/tests/unit/test_graph.py b/tests/unit/test_graph.py index 62fee7d..32f4bb5 100644 --- a/tests/unit/test_graph.py +++ b/tests/unit/test_graph.py @@ -94,6 +94,19 @@ def test_unknown_need(self): with pytest.raises(GraphValidationError, match="unknown task"): validate_graph(graph, {"a"}) + def test_non_list_needs(self): + from skitter.graph_gen import GraphValidationError, validate_graph + + # LLM sometimes emits a bare string instead of a list; iterating it would + # otherwise produce misleading per-character "unknown task" errors. + graph = { + "tasks": [ + {"id": "t1", "agent": "a", "needs": "t0", "terminal": True}, + ] + } + with pytest.raises(GraphValidationError, match="non-list"): + validate_graph(graph, {"a"}) + def test_terminal_has_dependents(self): from skitter.graph_gen import GraphValidationError, validate_graph diff --git a/tests/unit/test_llm.py b/tests/unit/test_llm.py index 6e6426a..f06ad3e 100644 --- a/tests/unit/test_llm.py +++ b/tests/unit/test_llm.py @@ -195,23 +195,6 @@ async def test_complete_openai_completions_with_system(self): assert messages[0] == {"role": "system", "content": "be helpful"} assert messages[1] == {"role": "user", "content": "hello"} - @pytest.mark.asyncio - async def test_complete_deepseek_uses_chat_api_and_default_base_url(self): - from skitter.config import LLMConfig - from skitter.llm import DEEPSEEK_BASE_URL, complete - - mock_client, mock_create = self._mock_openai_chat("deepseek response") - cfg = LLMConfig(model="deepseek-chat", api="deepseek", api_key="test-key") - with ( - patch("openai.AsyncOpenAI", return_value=mock_client) as mock_cls, - patch("skitter.llm.load_config", return_value=MagicMock(llm=cfg)), - ): - result = await complete("hello", model="deepseek-chat") - - assert result == "deepseek response" - assert mock_cls.call_args.kwargs["base_url"] == DEEPSEEK_BASE_URL - mock_create.assert_called_once() - @pytest.mark.asyncio async def test_complete_none_content_raises(self): from skitter.config import LLMConfig diff --git a/tests/unit/test_setup.py b/tests/unit/test_setup.py index a5fff09..2987bfd 100644 --- a/tests/unit/test_setup.py +++ b/tests/unit/test_setup.py @@ -43,19 +43,6 @@ def test_respects_custom_api(self): assert result is not None assert result["api"] == "openai" - def test_deepseek_defaults_to_deepseek_chat(self): - from skitter.setup import _collect_llm - - env = { - "SKITTER_LLM_API": "deepseek", - "SKITTER_LLM_API_KEY": "sk-deepseek", - } - with patch.dict("os.environ", env, clear=True): - result = _collect_llm(non_interactive=True, standalone=False, existing={}) - assert result is not None - assert result["api"] == "deepseek" - assert result["model"] == "deepseek-chat" - class TestSetupVerifyPassesLLMConfig: """_verify must pass LLM config to check() without mutating os.environ.""" From f7f492fc1b990ae99b670122439d7e3b3fb95267 Mon Sep 17 00:00:00 2001 From: YuShifan <894402575bt@gmail.com> Date: Mon, 29 Jun 2026 23:26:40 +0800 Subject: [PATCH 3/3] refactor: drop rigid "use every selected agent" graph constraint generate_graph no longer forces every provided agent into the graph: validate_graph loses the required_agent_ids check and the parameter, and the prompt now asks the model to use only the agents relevant to the instructions. This removes a brittle failure mode where the NL planner (which defaults to all available agents) could be forced to build a graph around agents unrelated to the request, failing generation outright. Also drop a stale apps/* pattern from .gitignore (frontend lives in web/, which has its own .gitignore; the top-level node_modules/ catch-all stays). --- .gitignore | 3 -- skitter/graph_gen.py | 17 ++-------- skitter/runtime_api.py | 2 +- tests/test_e2e.py | 2 +- tests/unit/test_graph.py | 67 ---------------------------------------- 5 files changed, 4 insertions(+), 87 deletions(-) diff --git a/.gitignore b/.gitignore index e6cc06d..e811d26 100644 --- a/.gitignore +++ b/.gitignore @@ -23,9 +23,6 @@ tests/claude-state/ # Frontend node_modules/ -apps/*/node_modules/ -apps/*/dist/ -apps/*/.vite/ # IDE .idea/ diff --git a/skitter/graph_gen.py b/skitter/graph_gen.py index f5d3edb..f1e82bd 100644 --- a/skitter/graph_gen.py +++ b/skitter/graph_gen.py @@ -26,8 +26,7 @@ Rules: - Every agent reference must match an agent ID from the provided list. -- Use every provided agent exactly once unless the instructions explicitly say - not to use a provided agent. +- Use the agents relevant to the instructions; you need not use every provided agent. - The graph must be a DAG (no cycles). - Task IDs must be unique. - "needs" lists upstream dependencies (tasks whose results this task requires). @@ -58,8 +57,6 @@ class GraphValidationError(Exception): def validate_graph( graph: dict, valid_agent_ids: set[str], - *, - required_agent_ids: set[str] | None = None, ) -> None: """Validate an orchestration graph. Raises GraphValidationError on failure.""" tasks = graph.get("tasks") @@ -92,15 +89,6 @@ def validate_graph( if need not in all_ids: raise GraphValidationError(f"Task '{tid}' needs unknown task '{need}'") - if required_agent_ids: - used_agent_ids = {str(t.get("agent", "")) for t in tasks} - missing_agent_ids = required_agent_ids - used_agent_ids - if missing_agent_ids: - raise GraphValidationError( - "Graph must use every selected agent; missing: " - f"{', '.join(sorted(missing_agent_ids))}" - ) - # Check for cycles using DFS on dependency edges adj: dict[str, list[str]] = {t.get("id"): t.get("needs", []) for t in tasks} visited: set[str] = set() @@ -147,7 +135,6 @@ async def generate_graph( agent_cards: dict[str, dict], *, model: str = "", - required_agent_ids: set[str] | None = None, ) -> dict: """Use LLM to generate an orchestration graph from instructions + agent cards. @@ -173,7 +160,7 @@ async def generate_graph( raise GraphValidationError(f"LLM returned invalid JSON: {e}") from e try: - validate_graph(graph, valid_ids, required_agent_ids=required_agent_ids) + validate_graph(graph, valid_ids) return graph except GraphValidationError as e: if attempt == 0: diff --git a/skitter/runtime_api.py b/skitter/runtime_api.py index 8ba6ba3..7cb19e9 100644 --- a/skitter/runtime_api.py +++ b/skitter/runtime_api.py @@ -618,7 +618,7 @@ async def _handle_create_app( # Generate orchestration graph via LLM try: - graph = await generate_graph(instructions, cards, required_agent_ids=set(cards)) + graph = await generate_graph(instructions, cards) except GraphValidationError as e: return ErrorResult(f"Graph generation failed: {e}") except Exception as e: diff --git a/tests/test_e2e.py b/tests/test_e2e.py index ff266c1..7d6a542 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -135,7 +135,7 @@ def mock_graph(): """Fixture to set the graph that generate_graph will return.""" container: dict = {} - async def _mock(instructions, agent_cards, *, model="", required_agent_ids=None): + async def _mock(instructions, agent_cards, *, model=""): return dict(container["graph"]) with patch("skitter.runtime_api.generate_graph", new=_mock): diff --git a/tests/unit/test_graph.py b/tests/unit/test_graph.py index 32f4bb5..1fed242 100644 --- a/tests/unit/test_graph.py +++ b/tests/unit/test_graph.py @@ -119,21 +119,6 @@ def test_terminal_has_dependents(self): with pytest.raises(GraphValidationError, match="must not have dependents"): validate_graph(graph, {"a", "b"}) - def test_required_agent_ids_must_be_used(self): - from skitter.graph_gen import GraphValidationError, validate_graph - - graph = { - "tasks": [ - {"id": "read", "agent": "reader", "needs": [], "terminal": True}, - ] - } - with pytest.raises(GraphValidationError, match="missing: analyzer"): - validate_graph( - graph, - {"reader", "analyzer"}, - required_agent_ids={"reader", "analyzer"}, - ) - class TestGraphGeneration: def _make_cards(self): @@ -237,58 +222,6 @@ async def test_generate_retries_on_validation_error(self): assert mock_llm.call_count == 2 assert graph["tasks"][0]["agent"] == "agent-reader-001" - @pytest.mark.asyncio - async def test_generate_retries_when_required_agent_missing(self): - from skitter.graph_gen import generate_graph - - missing_analyzer = json.dumps( - { - "tasks": [ - { - "id": "read", - "agent": "agent-reader-001", - "description": "Read", - "needs": [], - "terminal": True, - } - ] - } - ) - complete_graph = json.dumps( - { - "tasks": [ - { - "id": "read", - "agent": "agent-reader-001", - "description": "Read", - "needs": [], - }, - { - "id": "analyze", - "agent": "agent-analyzer-002", - "description": "Analyze", - "needs": ["read"], - "terminal": True, - }, - ] - } - ) - - with patch("skitter.graph_gen.complete", new_callable=AsyncMock) as mock_llm: - mock_llm.side_effect = [missing_analyzer, complete_graph] - graph = await generate_graph( - "Read and analyze", - self._make_cards(), - model="test", - required_agent_ids={"agent-reader-001", "agent-analyzer-002"}, - ) - - assert mock_llm.call_count == 2 - assert {task["agent"] for task in graph["tasks"]} == { - "agent-reader-001", - "agent-analyzer-002", - } - @pytest.mark.asyncio async def test_generate_fails_after_retries(self):