[eric] settings-agent/skills: prove live stdio guard on a real session, install->usable pipeline, and select->send threading

backend/tests/test_settings_meta_stdio_live.py

@@ -0,0 +1,115 @@
+"""Definitive proof: a SettingsWrite from the REAL stdio MCP server, against a
+REAL running backend, on a REAL session whose model is powered by a specific
+API key, refuses to clear THAT key but clears a different provider's key.
+
+Earlier coverage either drove the endpoint with an injected session (skipping the
+stdio subprocess) or drove the stdio subprocess with no session (hitting the
+fail-safe). This closes the gap: it boots uvicorn in-process (so the subprocess's
+HTTP call lands on a backend whose `agent_manager.sessions` we can populate),
+then runs `settings_meta_server.py` exactly as an agent would. No model required,
+the guard is provider-routing logic, not an LLM call.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import secrets
+import socket
+import subprocess
+import sys
+import threading
+import time
+
+import pytest
+import uvicorn
+
+from backend.main import app
+
+SERVER = os.path.join(os.path.dirname(os.path.dirname(__file__)), "apps", "agents", "settings_meta_server.py")
+
+
+def _free_port() -> int:
+    s = socket.socket()
+    s.bind(("127.0.0.1", 0))
+    port = s.getsockname()[1]
+    s.close()
+    return port
+
+
+@pytest.fixture
+def live_backend():
+    import backend.auth as auth_mod
+    if not auth_mod._TOKEN:
+        auth_mod._TOKEN = secrets.token_urlsafe(32)
+    port = _free_port()
+    server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error"))
+    thread = threading.Thread(target=server.run, daemon=True)
+    thread.start()
+    for _ in range(200):
+        if getattr(server, "started", False):
+            break
+        time.sleep(0.05)
+    assert getattr(server, "started", False), "uvicorn did not start"
+    yield port, auth_mod._TOKEN
+    server.should_exit = True
+    thread.join(timeout=5)
+
+
+@pytest.fixture
+def reset_settings():
+    from backend.apps.settings.settings import load_settings, _save_settings
+    original = load_settings().model_copy(deep=True)
+    yield
+    _save_settings(original)
+
+
+def _run_stdio(port: int, token: str, session_id: str, changes: dict) -> str:
+    env = {
+        **os.environ,
+        "OPENSWARM_PORT": str(port),
+        "OPENSWARM_AUTH_TOKEN": token,
+        "OPENSWARM_PARENT_SESSION_ID": session_id,
+    }
+    rpc = "\n".join([
+        json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}),
+        json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
+                    "params": {"name": "SettingsWrite", "arguments": {"changes": changes}}}),
+    ]) + "\n"
+    proc = subprocess.run([sys.executable, SERVER], input=rpc, capture_output=True, text=True, env=env, timeout=30)
+    msgs = [json.loads(l) for l in proc.stdout.splitlines() if l.strip()]
+    resp = next(m for m in msgs if m.get("id") == 2)
+    return resp["result"]["content"][0]["text"]
+
+
+def test_live_stdio_settingswrite_refuses_live_key_clears_other(live_backend, reset_settings):
+    port, token = live_backend
+    from backend.apps.settings.settings import load_settings, _save_settings
+    from backend.apps.agents.agent_manager import agent_manager
+    from backend.apps.agents.core.models import AgentSession
+
+    # A real run on opus-4-8 in own_key mode: the Anthropic key powers it; an
+    # OpenAI key is also connected (the "other provider").
+    s = load_settings()
+    s.connection_mode = "own_key"
+    s.anthropic_api_key = "sk-ant-LIVE-do-not-clear"
+    s.openai_api_key = "sk-oai-OTHER-ok-to-clear"
+    _save_settings(s)
+    agent_manager.sessions["live-stdio-test"] = AgentSession(id="live-stdio-test", name="t", model="opus-4-8")
+
+    try:
+        text = _run_stdio(port, token, "live-stdio-test",
+                          {"anthropic_api_key": "", "openai_api_key": "", "theme": "light"})
+    finally:
+        agent_manager.sessions.pop("live-stdio-test", None)
+
+    # The tool's own rendered result, exactly what the agent would read.
+    assert "Refused anthropic_api_key" in text, text
+    assert "powering this run" in text
+    assert "Applied" in text and "theme" in text
+
+    # And the truth on disk: live key kept, other cleared, benign applied.
+    final = load_settings()
+    assert final.anthropic_api_key == "sk-ant-LIVE-do-not-clear", "live key was cleared!"
+    assert not final.openai_api_key, "the other provider's key should have been cleared"
+    assert final.theme == "light"

backend/tests/test_settings_select_and_send.py

@@ -0,0 +1,53 @@
+"""Proof that selecting a Settings row and sending threads selected_setting_ids
+from the HTTP boundary into the run, and produces the focused context block.
+
+Covers the backend half definitively (HTTP -> send_message kwarg; the context
+builder output). The browser half (click a row -> the POST body carries
+selected_setting_ids) is proven live via CDP.
+"""
+
+from __future__ import annotations
+
+import secrets
+
+import pytest
+from fastapi.testclient import TestClient
+
+from backend.main import app
+
+
+@pytest.fixture
+def client():
+    import backend.auth as auth_mod
+    if not auth_mod._TOKEN:
+        auth_mod._TOKEN = secrets.token_urlsafe(32)
+    return TestClient(app, headers={"Authorization": f"Bearer {auth_mod._TOKEN}"})
+
+
+def test_message_endpoint_threads_selected_setting_ids(client, monkeypatch):
+    from backend.apps.agents.agent_manager import agent_manager
+    captured: dict = {}
+
+    async def fake_send(session_id, prompt, **kwargs):
+        captured["session_id"] = session_id
+        captured["kwargs"] = kwargs
+
+    monkeypatch.setattr(agent_manager, "send_message", fake_send)
+    r = client.post(
+        "/api/agents/sessions/sess-x/message",
+        json={"prompt": "flip my theme to light", "selected_setting_ids": ["theme", "default_model"]},
+    )
+    assert r.status_code == 200, r.text
+    assert captured["kwargs"].get("selected_setting_ids") == ["theme", "default_model"]
+
+
+def test_selected_settings_context_block_targets_the_fields():
+    from backend.apps.agents.manager.prompt.prompt_context import _build_selected_settings_context
+    assert _build_selected_settings_context(None) is None
+    assert _build_selected_settings_context([]) is None
+    block = _build_selected_settings_context(["theme", "default_model"])
+    assert "theme" in block and "default_model" in block
+    # It tells the agent to use the always-on settings tools on exactly these fields.
+    assert "SettingsRead" in block and "SettingsWrite" in block
+    # Targeting aid, not a gate: it focuses, never claims to unlock anything.
+    assert "Leave unrelated settings alone" in block

backend/tests/test_skill_registry_community.py

@@ -139,6 +139,47 @@ def test_install_dedups_instead_of_clobbering_existing_skill(skills_dir):
     assert {"pdf", "pdf-2"} <= ids
 
 
+def test_confirm_install_writes_folder_lists_and_injects(skills_dir, monkeypatch):
+    """End-to-end install->usable: confirm=true through the real /install endpoint
+    writes the folder skill, it shows up in /api/skills/list with supporting
+    files, and _resolve_attached_skills injects it with the folder path so an
+    agent can read its scripts. (resolve is mocked to skip the network; the live
+    GitHub resolve is proven separately.)"""
+    import secrets as _secrets
+    from fastapi.testclient import TestClient
+    from backend.main import app
+    from backend.apps.agents.manager.prompt.prompt_context import _resolve_attached_skills
+    import backend.auth as auth_mod
+    if not auth_mod._TOKEN:
+        auth_mod._TOKEN = _secrets.token_urlsafe(32)
+    client = TestClient(app, headers={"Authorization": f"Bearer {auth_mod._TOKEN}"})
+
+    async def fake_resolve(source, skill_id):
+        return {
+            "name": "PDF Tools", "description": "work with pdfs", "repo_url": "https://github.com/o/r",
+            "skill_id": skill_id,
+            "files": {"SKILL.md": "# PDF Tools\nRun scripts/extract.py to pull text.",
+                      "scripts/extract.py": "print('extract')"},
+            "scripts": ["scripts/extract.py"], "secret_findings": [],
+        }
+    monkeypatch.setattr("backend.apps.skill_registry.skill_registry.resolve_community_skill", fake_resolve)
+
+    r = client.post("/api/skill-registry/install", json={"source": "o/r", "skill_id": "pdf-tools", "confirm": True})
+    assert r.status_code == 200 and r.json()["installed"] is True
+    slug = r.json()["skill"]["id"]
+
+    # Listed via the real skills API, flagged as multi-file.
+    listed = {s["id"]: s for s in client.get("/api/skills/list").json()["skills"]}
+    assert slug in listed and listed[slug]["has_supporting_files"] is True
+    # On disk as a folder with the script.
+    assert (skills_dir / slug / "SKILL.md").exists()
+    assert (skills_dir / slug / "scripts" / "extract.py").exists()
+    # Injectable: the agent gets the body AND a pointer to the folder for on-demand reads.
+    block = _resolve_attached_skills([{"id": slug, "name": "PDF Tools", "content": "# PDF Tools\nRun scripts/extract.py to pull text."}])
+    assert "[Using skill: PDF Tools]" in block
+    assert str(skills_dir / slug) in block
+
+
 def test_write_folder_skill_blocks_path_traversal(skills_dir):
     skills_mod.write_folder_skill(
         "evil",