feat: natural-language orchestration, DeepSeek provider, coordinator hardening

.env.example

@@ -18,7 +18,7 @@ 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, or openai-completions.
 # SKITTER_LLM_API=anthropic
 
 # Model name (no provider prefix needed).

.gitignore

@@ -21,6 +21,9 @@ data/
 tests/workspace/
 tests/claude-state/
 
+# Frontend
+node_modules/
+
 # IDE
 .idea/
 .vscode/

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

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 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

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)]

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)

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 ---