diff --git a/skitter/agent_runner.py b/skitter/agent_runner.py index e5f97cb..c55cace 100644 --- a/skitter/agent_runner.py +++ b/skitter/agent_runner.py @@ -129,7 +129,9 @@ def _build_cli_cmd( if resume_id: cmd.extend(["--resume", resume_id]) elif _PERMISSION_MODE == "bypassPermissions": - cmd.extend(["--dangerously-skip-permissions", "--settings", _SANDBOX_SETTINGS]) + cmd.extend( + ["--dangerously-skip-permissions", "--settings", _SANDBOX_SETTINGS] + ) else: cmd.extend( ["--permission-mode", _PERMISSION_MODE, "--settings", _SANDBOX_SETTINGS] diff --git a/skitter/config.py b/skitter/config.py index b70c558..145263f 100644 --- a/skitter/config.py +++ b/skitter/config.py @@ -45,8 +45,19 @@ def skills_dir() -> Path: def write_env_file(env_path: Path, env_vars: dict[str, str]) -> None: - """Write a .env file (KEY=value per line).""" - lines = [f"{k}={v}" for k, v in env_vars.items() if v] + """Write a .env file (KEY=value per line). + + Values containing ``$`` are single-quoted so docker compose does not + try to interpolate them as variable references. + """ + lines: list[str] = [] + for k, v in env_vars.items(): + if not v: + continue + if "$" in v: + # Single quotes prevent variable interpolation in docker compose + v = f"'{v}'" + lines.append(f"{k}={v}") env_path.write_text("\n".join(lines) + "\n" if lines else "") diff --git a/skitter/coordinator/models.py b/skitter/coordinator/models.py index 9252552..6811058 100644 --- a/skitter/coordinator/models.py +++ b/skitter/coordinator/models.py @@ -16,6 +16,7 @@ class SessionTask: target: TaskTarget | None = None dispatch_correlation: str = "" # MQTT Correlation Data sent with dispatch dispatch_task_id: str = "" # A2A Task.id sent to agent; used for CancelTask + reply_topic: str = "" # MQTT reply topic for unsubscribe after completion @dataclass diff --git a/skitter/coordinator/reply_handler.py b/skitter/coordinator/reply_handler.py index bb78ed3..edf962f 100644 --- a/skitter/coordinator/reply_handler.py +++ b/skitter/coordinator/reply_handler.py @@ -118,6 +118,11 @@ async def complete_task( state.results[node_id] = result state.inflight.discard(node_id) + # Free broker subscription slot + task = state.graph.get(node_id) + if task and task.reply_topic: + await coord._unsubscribe_reply(task.reply_topic) + # Update DB db_task_row_id = f"{state.session_id}/{node_id}" await coord._adb.update_task( @@ -145,6 +150,11 @@ async def fail_task( state.inflight.discard(node_id) state.failed.add(node_id) + # Free broker subscription slot + task = state.graph.get(node_id) + if task and task.reply_topic: + await coord._unsubscribe_reply(task.reply_topic) + # Update DB db_task_row_id = f"{state.session_id}/{node_id}" await coord._adb.update_task( diff --git a/skitter/coordinator/service.py b/skitter/coordinator/service.py index ca48bc3..ccd0856 100644 --- a/skitter/coordinator/service.py +++ b/skitter/coordinator/service.py @@ -96,6 +96,16 @@ def _clear_context_active(self, state: SessionState) -> None: if key[1] and self._context_active.get(key) == state.session_id: del self._context_active[key] + async def _unsubscribe_reply(self, reply_topic: str) -> None: + """Unsubscribe from a task reply topic to free broker subscription slots.""" + if reply_topic and reply_topic in self._reply_subscriptions: + self._reply_subscriptions.discard(reply_topic) + if self._client: + try: + await self._client.unsubscribe(reply_topic) + except Exception: + pass + # --- Session events --- async def _publish_event( @@ -292,6 +302,7 @@ async def _dispatch_task(self, state: SessionState, node_id: str) -> None: if self._client and reply_t not in self._reply_subscriptions: await self._client.subscribe(reply_t, qos=1) self._reply_subscriptions.add(reply_t) + task.reply_topic = reply_t a2a_req = A2ARequest( text=prompt, request_id=correlation, @@ -437,6 +448,9 @@ async def _cancel_session_cleanup(self, session_id: str) -> None: log.warning("Failed to send CancelTask for %s/%s", session_id, tid) for tid in state.pending | state.inflight: + task_def = state.graph.get(tid) + if task_def and task_def.reply_topic: + await self._unsubscribe_reply(task_def.reply_topic) await self._adb.update_task( f"{session_id}/{tid}", state=TaskState.CANCELED, completed_at=now ) @@ -966,6 +980,9 @@ def _parse_agent_id_from_topic(topic: str) -> str: def main() -> None: + from skitter.config import configure_logging + + configure_logging() db = open_db() coord = Coordinator(db) try: diff --git a/skitter/create_agent.py b/skitter/create_agent.py index 13150ae..5e69fc4 100644 --- a/skitter/create_agent.py +++ b/skitter/create_agent.py @@ -8,6 +8,7 @@ """ import argparse +import getpass import logging import os import re @@ -105,8 +106,8 @@ def _prompt_auth(runtime: str) -> dict[str, str]: label = f"{var}" if hint: label += f" ({hint})" - prompt_str = f"{label} [{default[:8]}...]" if default else label - value = input(f"{prompt_str}: ").strip() or default + prompt_str = f"{label} [{default[:4]}{'*' * 4}...]" if default else label + value = getpass.getpass(f"{prompt_str}: ").strip() or default if value: result[var] = value break # use the first available credential diff --git a/skitter/services.py b/skitter/services.py index f00aba5..ff9ecc3 100644 --- a/skitter/services.py +++ b/skitter/services.py @@ -290,6 +290,15 @@ def _generate_compose( if volumes: compose["volumes"] = volumes + # Escape $ as $$ in environment values so docker compose doesn't + # interpret them as variable references (e.g. passwords with $). + for svc in compose["services"].values(): + if "environment" in svc: + svc["environment"] = { + k: v.replace("$", "$$") if isinstance(v, str) else v + for k, v in svc["environment"].items() + } + header = "# Auto-generated by 'skitter up'. Do not edit.\n" return header + yaml.dump(compose, default_flow_style=False, sort_keys=False) @@ -356,7 +365,7 @@ def up(argv: list[str] | None = None) -> None: targets = ["emqx"] elif single_agent: targets = [f"agent-{single_agent}"] - _compose("up", "-d", "--wait", *targets) + _compose("up", "-d", "--wait", "--remove-orphans", *targets) if broker_only: print("Broker ready. Use 'uv run skitter' to run coordinator from source.") @@ -512,7 +521,9 @@ def status(argv: list[str] | None = None) -> None: # Agent summary if agents: - running_names = {n for n, s, _ in containers if s in ("running", "Up")} + running_names = { + n for n, s, _ in containers if "running" in s.lower() or "Up" in s + } running_agents = { aid for aid, _, _ in agents if _agent_container_name(aid) in running_names } diff --git a/skitter/setup.py b/skitter/setup.py index c33b573..6666d7b 100644 --- a/skitter/setup.py +++ b/skitter/setup.py @@ -5,6 +5,7 @@ """ import asyncio +import getpass import logging import os import sys @@ -17,13 +18,21 @@ log = logging.getLogger("skitter.setup") -def _prompt(msg: str, default: str = "") -> str: - suffix = f" [{default}]" if default else "" +def _prompt(msg: str, default: str = "", *, secret: bool = False) -> str: + if secret: + hint = " [****]" if default else "" + else: + hint = f" [{default}]" if default else "" try: - value = input(f"{msg}{suffix}: ").strip() + if secret: + value = getpass.getpass(f"{msg}{hint}: ").strip() + else: + value = input(f"{msg}{hint}: ").strip() except (EOFError, KeyboardInterrupt): print("\nAborted.", file=sys.stderr) sys.exit(1) + if value == "-": + return "" return value or default @@ -114,9 +123,10 @@ def _collect_llm( model = _prompt("Model", existing.get("model", default_models.get(api, ""))) - api_key = os.environ.get("SKITTER_LLM_API_KEY", "") or existing.get("api_key", "") - if not api_key: - api_key = _prompt("API key") + default_key = os.environ.get("SKITTER_LLM_API_KEY", "") or existing.get( + "api_key", "" + ) + api_key = _prompt("API key", default_key, secret=True) if api_key: os.environ["SKITTER_LLM_API_KEY"] = api_key @@ -173,7 +183,7 @@ def _collect_broker(non_interactive: bool, existing: dict) -> dict: existing.get("url", ""), ) username = _prompt("Username", existing.get("username", "")) - password = _prompt("Password", existing.get("password", "")) + password = _prompt("Password", existing.get("password", ""), secret=True) ca_cert = _prompt("CA cert path (optional)", existing.get("ca_cert", "")) return { "tier": "serverless", @@ -194,7 +204,7 @@ def _collect_broker(non_interactive: bool, existing: dict) -> dict: "Consider the 'docker' tier or use an externally routable address." ) username = _prompt("Username (optional)", existing.get("username", "")) - password = _prompt("Password (optional)", existing.get("password", "")) + password = _prompt("Password (optional)", existing.get("password", ""), secret=True) return { "tier": "custom", "url": url, diff --git a/tests/unit/test_a2a.py b/tests/unit/test_a2a.py index f0c1444..1387a64 100644 --- a/tests/unit/test_a2a.py +++ b/tests/unit/test_a2a.py @@ -1100,7 +1100,11 @@ def test_namespace_resolves_from_config(self, tmp_path): with patch.dict( "os.environ", - {"SKITTER_HOME": str(tmp_path)}, + { + "SKITTER_HOME": str(tmp_path), + "SKITTER_A2A_ORG": "", + "SKITTER_A2A_UNIT": "", + }, clear=False, ): org = a2a_mod.a2a_org() @@ -1120,7 +1124,11 @@ def test_topic_builders_use_lazy_namespace(self, tmp_path): with patch.dict( "os.environ", - {"SKITTER_HOME": str(tmp_path)}, + { + "SKITTER_HOME": str(tmp_path), + "SKITTER_A2A_ORG": "", + "SKITTER_A2A_UNIT": "", + }, clear=False, ): topic = a2a_mod.topic_request("test-agent") diff --git a/tests/unit/test_services.py b/tests/unit/test_services.py index d984636..6558ab2 100644 --- a/tests/unit/test_services.py +++ b/tests/unit/test_services.py @@ -130,7 +130,13 @@ def test_load_config_uses_skitter_home(self, tmp_path): cfg_file.write_text("broker:\n tier: public\n url: mqtt://example.com:1883\n") with patch.dict( "os.environ", - {"SKITTER_HOME": str(tmp_path), "SKITTER_LLM_MODEL": ""}, + { + "SKITTER_HOME": str(tmp_path), + "SKITTER_LLM_MODEL": "", + "MQTT_BROKER_URL": "", + "MQTT_USERNAME": "", + "MQTT_PASSWORD": "", + }, clear=False, ): cfg = load_config()