[TH-9519] fix(harness): ask for the ElevenLabs key under the name LiveKit's plugin reads

oss/simulation-acceptance/.env.example

@@ -32,6 +32,7 @@ GOOGLE_CLOUD_PROJECT=
 DEEPGRAM_API_KEY=
 # CARTESIA_API_KEY=
 # OPENAI_API_KEY=
+# ELEVEN_API_KEY=
 # ELEVENLABS_API_KEY=
 
 # LiveKit web target. Use a concrete company name in the prompt; do not leave

src/fi/alk/harness/credentials.py

@@ -264,7 +264,7 @@ def _literal_compose_environment(content: str) -> set[str]:
     (("LIVEKIT_", "ACCEPTANCE_LIVEKIT_"), "livekit", "connect to the voice room"),
     (("DEEPGRAM_",), "deepgram", "speech-to-text"),
     (("CARTESIA_",), "cartesia", "text-to-speech"),
-    (("ELEVENLABS_",), "elevenlabs", "text-to-speech"),
+    (("ELEVENLABS_", "ELEVEN_"), "elevenlabs", "text-to-speech"),
     (("OPENAI_",), "openai", "language or realtime model"),
     (("ANTHROPIC_",), "anthropic", "language model"),
     (("GOOGLE_", "VERTEX_", "CLOUD_ML_"), "google_cloud", "model or cloud runtime"),
@@ -298,11 +298,70 @@ def _literal_compose_environment(content: str) -> set[str]:
 _LIVEKIT_PROVIDER_CREDENTIALS = {
     "deepgram": ("DEEPGRAM_API_KEY",),
     "cartesia": ("CARTESIA_API_KEY",),
-    "elevenlabs": ("ELEVENLABS_API_KEY",),
+    "elevenlabs": ("ELEVEN_API_KEY",),
     "openai": ("OPENAI_API_KEY",),
 }
 
 
+# When a constructor's ``api_key=`` can't be read, ElevenLabs' own code may read either name.
+_LIVEKIT_UNREADABLE_KEY_EXTRA = {"elevenlabs": ("ELEVENLABS_API_KEY",)}
+_CODE_ENV_NAME = re.compile(r"[A-Z][A-Z0-9_]{2,}")
+
+
+def _plugin_key_names(node: ast.Call, provider: str) -> tuple[str, ...]:
+    """Names a LiveKit plugin constructor needs, judged by the ``api_key=`` it passes.
+
+    Only errs towards asking: anything that can't be read keeps the plugin's own row.
+    """
+    own = _LIVEKIT_PROVIDER_CREDENTIALS[provider]
+    fallback = own + _LIVEKIT_UNREADABLE_KEY_EXTRA.get(provider, ())
+    keys = [keyword for keyword in node.keywords if keyword.arg == "api_key"]
+    if not keys:
+        return fallback if any(k.arg is None for k in node.keywords) else own
+    value = keys[0].value
+    if isinstance(value, ast.Constant) and isinstance(value.value, str):
+        return () if value.value.strip() else fallback
+    if (
+        isinstance(value, ast.Subscript)
+        and _dotted_name(value.value) == "os.environ"
+        and isinstance(value.slice, ast.Constant)
+        and isinstance(value.slice.value, str)
+        and _CODE_ENV_NAME.fullmatch(value.slice.value)
+    ):
+        return ()
+    if (
+        isinstance(value, ast.Call)
+        and _dotted_name(value.func) in {"os.getenv", "os.environ.get"}
+        and value.args
+        and not any(isinstance(arg, ast.Starred) for arg in value.args)
+        and not any(k.arg is None for k in value.keywords)
+        and len(value.args) <= 2
+        and isinstance(value.args[0], ast.Constant)
+        and isinstance(value.args[0].value, str)
+        and _CODE_ENV_NAME.fullmatch(value.args[0].value)
+    ):
+        name = value.args[0].value
+        if name in _NON_USER_CONFIGURATION or name.startswith("ALK_"):
+            # A name the customer can never supply. Some plugins then fall back to their own
+            # variable (``api_key or os.environ.get(...)``: Cartesia, Deepgram TTS, OpenAI
+            # realtime); others refuse an explicit None. Ask for the plugin's own name.
+            return fallback
+        default = value.args[1] if len(value.args) > 1 else None
+        for keyword in value.keywords:
+            if keyword.arg == "default" and default is None:
+                default = keyword.value
+        if default is None or (
+            isinstance(default, ast.Constant) and default.value is None
+        ):
+            return (name,)
+        if isinstance(default, ast.Constant) and isinstance(default.value, str):
+            text = default.value.strip()
+            if not text or _PLACEHOLDER_VALUE.fullmatch(text):
+                return (name,)
+            return ()
+    return fallback
+
+
 def _dotted_name(node: ast.AST) -> str:
     if isinstance(node, ast.Name):
         return node.id
@@ -393,7 +452,7 @@ def _python_sdk_requirements(content: str) -> list[tuple[str, str]]:
             "TTS",
             "RealtimeModel",
         }:
-            for name in _LIVEKIT_PROVIDER_CREDENTIALS[provider]:
+            for name in _plugin_key_names(node, provider):
                 found.add((name, f"sdk:livekit.plugins.{provider}"))
         if provider == "google" and constructor in {"LLM", "STT", "TTS"}:
             vertex = next(
@@ -602,6 +661,11 @@ def record(
         if path.suffix.lower() == ".py":
             for name, origin in _python_sdk_requirements(content):
                 record(name, required=True)
+                if name not in findings:
+                    # A reserved name record() leaves out. _plugin_key_names already asks for
+                    # the plugin's own name instead (getenv); this keeps any other form from
+                    # raising KeyError here.
+                    continue
                 detected = findings[name]["detected_from"]
                 assert isinstance(detected, set)
                 detected.add(origin)
@@ -625,7 +689,11 @@ def record(
     for name, finding in sorted(findings.items()):
         kind = _kind(name)
         required = bool(finding["required"])
-        if kind is RequirementKind.CONFIGURATION and finding["declared_default"]:
+        if (
+            kind is RequirementKind.CONFIGURATION
+            and finding["declared_default"]
+            and not any(str(o).startswith("sdk:") for o in finding["detected_from"])
+        ):
             required = False
         if kind is RequirementKind.HARNESS_INFRASTRUCTURE:
             status = RequirementStatus.HARNESS_PROVIDED

tests/test_harness_credentials.py

@@ -1069,3 +1069,179 @@ def test_a_walk_that_cant_parse_a_file_marks_the_manifest_truncated(tmp_path):
 
     manifest = discover_credentials(root, agent_subfolder="agents/a")
     assert manifest.truncated
+
+
+# --- LiveKit plugin api_key= handling (TH-9519) ---------------------------------------
+
+_LK_ENV = ["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"]
+
+
+def _plugin_agent(tmp_path: Path, plugin: str, call: str, *, pre: str = "") -> None:
+    _write(
+        tmp_path,
+        "agent.py",
+        f"import os\nfrom livekit.plugins import {plugin}\n{pre}\nx = {plugin}.{call}\n",
+    )
+
+
+def _lk(tmp_path: Path, *supplied: str):
+    manifest = discover_credentials(
+        tmp_path, provided_environment=[*_LK_ENV, *supplied]
+    )
+    return manifest, _requirements(manifest)
+
+
+def _missing(tmp_path: Path, *supplied: str) -> set[str]:
+    manifest, found = _lk(tmp_path, *supplied)
+    assert manifest.ready is bool(
+        not any(r.status is RequirementStatus.MISSING for r in found.values())
+    )
+    return {
+        name for name, r in found.items() if r.status is RequirementStatus.MISSING
+    }
+
+
+@pytest.mark.parametrize("call", ["TTS()", "STT()"])
+def test_elevenlabs_plugin_asks_for_the_name_the_plugin_reads(tmp_path, call):
+    _plugin_agent(tmp_path, "elevenlabs", call)
+    _, found = _lk(tmp_path)
+    assert "ELEVENLABS_API_KEY" not in found
+    assert found["ELEVEN_API_KEY"].provider == "elevenlabs"
+    assert found["ELEVEN_API_KEY"].status is RequirementStatus.MISSING
+    manifest, _ = _lk(tmp_path, "ELEVEN_API_KEY")
+    assert manifest.ready
+
+
+def test_elevenlabs_environment_index_is_read_as_that_name_only(tmp_path):
+    _plugin_agent(
+        tmp_path, "elevenlabs", 'TTS(api_key=os.environ["ELEVENLABS_API_KEY"])'
+    )
+    assert _missing(tmp_path) == {"ELEVENLABS_API_KEY"}
+    _, found = _lk(tmp_path)
+    assert "ELEVEN_API_KEY" not in found
+    assert _lk(tmp_path, "ELEVENLABS_API_KEY")[0].ready
+
+
+@pytest.mark.parametrize(
+    "plugin,call,read_name,own_name",
+    [
+        ("elevenlabs", 'TTS(api_key=os.getenv("ELEVENLABS_API_KEY"))', "ELEVENLABS_API_KEY", "ELEVEN_API_KEY"),
+        ("elevenlabs", 'TTS(api_key=os.environ.get("ELEVENLABS_API_KEY", ""))', "ELEVENLABS_API_KEY", "ELEVEN_API_KEY"),
+        ("deepgram", 'STT(api_key=os.getenv("MY_DEEPGRAM"))', "MY_DEEPGRAM", "DEEPGRAM_API_KEY"),
+        ("deepgram", 'STT(api_key=os.environ.get("DG_KEY", ""))', "DG_KEY", "DEEPGRAM_API_KEY"),
+        ("deepgram", 'STT(api_key=os.getenv("DG_KEY", None))', "DG_KEY", "DEEPGRAM_API_KEY"),
+        ("deepgram", 'STT(api_key=os.getenv("DG_KEY", default=""))', "DG_KEY", "DEEPGRAM_API_KEY"),
+        ("elevenlabs", 'TTS(api_key=os.getenv("ELEVENLABS_API_KEY", "your-api-key"))', "ELEVENLABS_API_KEY", "ELEVEN_API_KEY"),
+    ],
+)
+def test_a_nullable_read_passed_as_api_key_is_required_by_its_own_name(
+    tmp_path, plugin, call, read_name, own_name
+):
+    _plugin_agent(tmp_path, plugin, call)
+    assert _missing(tmp_path) == {read_name}
+    _, found = _lk(tmp_path)
+    assert own_name not in found
+    assert found[read_name].required is True
+    assert _lk(tmp_path, read_name)[0].ready
+
+
+@pytest.mark.parametrize(
+    "plugin,call",
+    [
+        ("deepgram", 'STT(api_key=os.getenv("X_KEY", "fallback-value"))'),
+        ("deepgram", 'STT(api_key=os.getenv("DG_KEY", default="fallback-value"))'),
+        ("deepgram", 'STT(api_key="literal")'),
+        ("elevenlabs", 'TTS(api_key="literal")'),
+    ],
+)
+def test_a_literal_or_defaulted_api_key_adds_no_requirement(tmp_path, plugin, call):
+    _plugin_agent(tmp_path, plugin, call)
+    manifest, found = _lk(tmp_path)
+    assert manifest.ready
+    assert not any(
+        n in found for n in ("DEEPGRAM_API_KEY", "ELEVEN_API_KEY", "ELEVENLABS_API_KEY")
+    )
+    assert not any(
+        o.startswith("sdk:") for r in found.values() for o in r.detected_from
+    )
+
+
+@pytest.mark.parametrize(
+    "call,pre",
+    [
+        ("STT(api_key=KEY)", 'KEY = os.getenv("DEEPGRAM_API_KEY")'),
+        ("STT(api_key=settings.deepgram_api_key)", ""),
+        ("STT(api_key=os.getenv('DEEPGRAM_API_KEY') or '')", ""),
+        ("STT(**opts)", ""),
+        ("STT(api_key='')", ""),
+        ('STT(api_key=os.environ["dg_key"])', ""),
+        ('STT(api_key=os.getenv("dg_key"))', ""),
+        ('STT(api_key="  ")', ""),
+    ],
+)
+def test_an_unreadable_api_key_keeps_the_plugins_own_requirement(tmp_path, call, pre):
+    _plugin_agent(tmp_path, "deepgram", call, pre=pre)
+    assert "DEEPGRAM_API_KEY" in _missing(tmp_path)
+    assert _lk(tmp_path, "DEEPGRAM_API_KEY")[0].ready
+
+
+@pytest.mark.parametrize(
+    "plugin, call, own_name",
+    [
+        ("openai", 'LLM(api_key=os.getenv("ALK_CLAUDE_GATEWAY_API_KEY"))', "OPENAI_API_KEY"),
+        ("deepgram", 'STT(api_key=os.getenv("HOST"))', "DEEPGRAM_API_KEY"),
+        ("cartesia", 'TTS(api_key=os.getenv("ALK_X"))', "CARTESIA_API_KEY"),
+        ("openai", 'realtime.RealtimeModel(api_key=os.getenv("ALK_X"))', "OPENAI_API_KEY"),
+    ],
+)
+def test_a_reserved_name_passed_as_the_key_asks_for_the_plugins_own_name(
+    tmp_path, plugin, call, own_name
+):
+    _plugin_agent(tmp_path, plugin, call)
+    _, found = _lk(tmp_path)
+    assert "ALK_CLAUDE_GATEWAY_API_KEY" not in found and "HOST" not in found
+    assert "ALK_X" not in found
+    # The customer can never supply the reserved name, and several plugins then read their own
+    # variable, so the check asks for that rather than calling the agent ready.
+    assert own_name in _missing(tmp_path)
+
+
+def test_a_bare_getenv_imported_from_os_keeps_the_plugins_own_requirement(tmp_path):
+    _write(
+        tmp_path,
+        "agent.py",
+        "from os import getenv\nfrom livekit.plugins import deepgram\n"
+        'x = deepgram.STT(api_key=getenv("DEEPGRAM_API_KEY"))\n',
+    )
+    assert "DEEPGRAM_API_KEY" in _missing(tmp_path)
+
+
+def test_elevenlabs_options_dict_is_not_ready_without_the_plugins_name(tmp_path):
+    _plugin_agent(tmp_path, "elevenlabs", "TTS(**opts)")
+    assert {"ELEVEN_API_KEY", "ELEVENLABS_API_KEY"} <= _missing(tmp_path)
+
+
+@pytest.mark.parametrize(
+    "call,pre",
+    [
+        ("TTS(api_key=KEY)", 'KEY = os.getenv("ELEVENLABS_API_KEY")'),
+        ("TTS(api_key=settings.elevenlabs_api_key)", ""),
+    ],
+)
+def test_an_unreadable_elevenlabs_api_key_asks_for_both_names(tmp_path, call, pre):
+    _plugin_agent(tmp_path, "elevenlabs", call, pre=pre)
+    assert "ELEVENLABS_API_KEY" in _missing(tmp_path, "ELEVEN_API_KEY")
+    assert not _lk(tmp_path, "ELEVEN_API_KEY")[0].ready
+    assert _lk(tmp_path, "ELEVEN_API_KEY", "ELEVENLABS_API_KEY")[0].ready
+
+
+def test_code_reading_the_eleven_names_is_labelled_elevenlabs(tmp_path):
+    _write(
+        tmp_path,
+        "agent.py",
+        'import os\na = os.environ["ELEVEN_API_KEY"]\nb = os.environ["ELEVENLABS_API_KEY"]\n',
+    )
+    _, found = _lk(tmp_path)
+    for name in ("ELEVEN_API_KEY", "ELEVENLABS_API_KEY"):
+        assert found[name].provider == "elevenlabs"
+        assert found[name].required is True