Add oída→germ akousma bridge (three buttons) over the shared store

- oida/akousma_bridge.py: build akousma from a listen, persist to the shared
  akousmata store, and mint germ deep links for the three buttons
  (open as sound / open as prompt / explore lineage). Lazy-imports FastAPI so it
  stays usable in tests.
- server.py: registers /germ/handoff (POST) and /germ/link (GET); optional so
  oída still boots without the akousma package. Fixed wildcard-bind message to
  name OIDA_AUTH_TOKEN.
- pyproject: akousma dependency via [tool.uv.sources] path to earworm/py-akousma.
- tests: bridge + cross-app round-trip (listen→A, germ child B, lineage A→B,
  algophony query). Full suite 107 tests pass.

Author: emeisazam

oida/akousma_bridge.py

@@ -0,0 +1,140 @@
+"""oída → germ bridge over the shared akousma protocol.
+
+oída is generative ears; germ is generative voice. After a listen, oída persists an
+**akousma** (the sound's memory record) into the shared **akousmata** store and hands germ
+an ``akousma_id`` via a deep link. The three UI buttons map to three modes:
+
+- ``sound``   — "open as sound":    load the listened fragment as an audio source in germ.
+- ``prompt``  — "open as prompt":   open the listening result as a generation prompt in germ.
+- ``lineage`` — "explore lineage":  open germ's genetic-ancestry explorer on this akousma.
+
+Requires the ``akousma`` package (earworm/packages/py-akousma).
+"""
+from __future__ import annotations
+
+import os
+from typing import Any
+from urllib.parse import urlencode
+
+import akousma
+
+MODES = ("sound", "prompt", "lineage")
+
+
+def germ_base_url() -> str:
+    """Germ's base URL for deep links (``OIDA_GERM_URL``, default local dashboard)."""
+    return os.getenv("OIDA_GERM_URL", "http://127.0.0.1:5178").rstrip("/")
+
+
+def germ_deep_link(akousma_id: str, mode: str) -> str:
+    if mode not in MODES:
+        raise ValueError(f"mode must be one of {MODES}, got {mode!r}")
+    query = urlencode({"akousma": akousma_id, "mode": mode})
+    return f"{germ_base_url()}/import?{query}"
+
+
+def _origin_to_source_type(origin: str) -> str:
+    """Map oída's capture origin to Earworm's provenance source_type vocabulary."""
+    return {
+        "live-input": "recorded",
+        "system-output": "recorded",
+        "file": "imported",
+        "generated": "generated",
+    }.get(origin, "unknown")
+
+
+def build_akousma_from_listen(
+    *,
+    audio: dict[str, Any],
+    listening: dict[str, Any] | None = None,
+    origin: str = "file",
+    device: str | None = None,
+    session_id: str | None = None,
+    tags: list[str] | None = None,
+) -> dict[str, Any]:
+    """Build a valid akousma record from an oída listen result.
+
+    ``audio`` needs at least ``asset_id`` (and ideally ``uri``/``content_hash``/duration).
+    ``listening`` is namespaced per producer, e.g. ``{"oida.signal": {...}, "akouo.describe": {...}}``.
+    """
+    record = akousma.new_akousma(
+        audio=audio,
+        originating_app="oida",
+        source_type=_origin_to_source_type(origin),
+        origin=origin,
+        listening=listening or {},
+        operation="listen",
+        tags=tags,
+        session_id=session_id,
+    )
+    if device:
+        record["provenance"]["device"] = device
+    return record
+
+
+def handoff_to_germ(
+    record: dict[str, Any],
+    mode: str,
+    *,
+    store: "akousma.AkousmataStore | None" = None,
+) -> dict[str, Any]:
+    """Persist ``record`` to the shared akousmata store and return the germ deep link.
+
+    Returns ``{"akousma_id", "mode", "germ_url"}``. Backing the three oída buttons.
+    """
+    if mode not in MODES:
+        raise ValueError(f"mode must be one of {MODES}, got {mode!r}")
+    owns_store = store is None
+    store = store or akousma.AkousmataStore()
+    try:
+        akousma_id = store.put(record)
+    finally:
+        if owns_store:
+            store.close()
+    return {"akousma_id": akousma_id, "mode": mode, "germ_url": germ_deep_link(akousma_id, mode)}
+
+
+def build_germ_router():
+    """FastAPI router backing the three oída→germ buttons. Imported lazily so this
+    module stays usable without FastAPI (e.g. in cross-app tests)."""
+    from fastapi import APIRouter, HTTPException
+    from pydantic import BaseModel
+
+    class GermHandoffRequest(BaseModel):
+        mode: str
+        audio: dict[str, Any]
+        listening: dict[str, Any] | None = None
+        origin: str = "file"
+        device: str | None = None
+        session_id: str | None = None
+        tags: list[str] | None = None
+
+    router = APIRouter(prefix="/germ", tags=["germ"])
+
+    @router.post("/handoff")
+    def germ_handoff(req: GermHandoffRequest) -> dict[str, Any]:
+        try:
+            record = build_akousma_from_listen(
+                audio=req.audio,
+                listening=req.listening,
+                origin=req.origin,
+                device=req.device,
+                session_id=req.session_id,
+                tags=req.tags,
+            )
+            return handoff_to_germ(record, req.mode)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @router.get("/link")
+    def germ_link(akousma_id: str, mode: str = "lineage") -> dict[str, Any]:
+        try:
+            return {
+                "akousma_id": akousma_id,
+                "mode": mode,
+                "germ_url": germ_deep_link(akousma_id, mode),
+            }
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    return router

oida/server.py

@@ -473,10 +473,18 @@ async def lifespan(_app: Any):
     static_dir = Path(__file__).resolve().parent / "static"
     app.mount("/static", StaticFiles(directory=static_dir), name="static")
 
+    # oída→germ bridge (three buttons) over the shared akousma store; optional.
+    try:
+        from .akousma_bridge import build_germ_router
+
+        app.include_router(build_germ_router())
+    except ImportError:
+        pass  # akousma package not installed; oída still boots without the bridge
+
     wildcard_bind = str(config.host) in {"0.0.0.0", "::", ""}
     if wildcard_bind and not config.auth_token:
         raise RuntimeError(
-            "Refusing to bind oida on a wildcard host without HMM_AUTH_TOKEN or AEAR_AUTH_TOKEN. "
+            "Refusing to bind oida on a wildcard host without OIDA_AUTH_TOKEN (or legacy HMM_/AEAR_AUTH_TOKEN). "
             "Use 127.0.0.1 for tokenless local operation."
         )
     if wildcard_bind:

pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">=3.12"
 license = { text = "Apache-2.0" }
 authors = [{ name = "Sonic Field Labs" }]
 dependencies = [
+  "akousma>=0.1.0",
   "fastapi>=0.115",
   "jsonschema>=4.22",
   "numpy>=1.26",
@@ -16,6 +17,9 @@ dependencies = [
   "uvicorn[standard]>=0.30",
 ]
 
+[tool.uv.sources]
+akousma = { path = "../earworm/packages/py-akousma", editable = true }
+
 [project.optional-dependencies]
 moss = [
   "torch>=2.9",

tests/test_akousma_bridge.py

@@ -0,0 +1,73 @@
+"""oída→germ bridge + cross-app akousma round-trip (Phase 4 acceptance)."""
+import tempfile
+import unittest
+
+import akousma
+from oida import akousma_bridge
+
+
+class TestGermDeepLinks(unittest.TestCase):
+    def test_deep_link_format(self):
+        url = akousma_bridge.germ_deep_link("akm_1", "prompt")
+        self.assertIn("/import?", url)
+        self.assertIn("akousma=akm_1", url)
+        self.assertIn("mode=prompt", url)
+
+    def test_rejects_unknown_mode(self):
+        with self.assertRaises(ValueError):
+            akousma_bridge.germ_deep_link("akm_1", "bogus")
+
+    def test_origin_maps_to_earworm_source_type(self):
+        rec = akousma_bridge.build_akousma_from_listen(
+            audio={"asset_id": "a1"}, origin="live-input"
+        )
+        self.assertEqual(rec["provenance"]["source_type"], "recorded")
+        self.assertEqual(rec["provenance"]["origin"], "live-input")
+        self.assertEqual(rec["provenance"]["originating_app"], "oida")
+
+
+class TestCrossAppRoundTrip(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.store = akousma.AkousmataStore(self.tmp.name)
+
+    def tearDown(self):
+        self.store.close()
+        self.tmp.cleanup()
+
+    def test_listen_to_generate_to_lineage(self):
+        # 1) oída listens to a file → akousma A, "open as prompt" hands it to germ.
+        a = akousma_bridge.build_akousma_from_listen(
+            audio={"asset_id": "file1", "uri": "akousmata://objects/x.wav", "duration_seconds": 8.0},
+            origin="file",
+            listening={"oida.signal": {"class": "tonal"}, "akouo.describe": {"summary": "struck bell"}},
+        )
+        handoff = akousma_bridge.handoff_to_germ(a, "prompt", store=self.store)
+        A = handoff["akousma_id"]
+        self.assertIn("mode=prompt", handoff["germ_url"])
+
+        # 2) germ generates a child B whose lineage points at A.
+        b = akousma.new_akousma(
+            audio={"asset_id": "gen1"},
+            originating_app="germ",
+            source_type="generated",
+            origin="generated",
+            parent_akousma_ids=[A],
+            operation="transform",
+            prompt="make it metallic",
+            model="stable-audio-3",
+        )
+        B = self.store.put(b)
+
+        # 3) germ's lineage explorer walks ancestry; "explore lineage" from oída shows A→B.
+        self.assertEqual(self.store.ancestors(B), [A])
+        self.assertEqual(self.store.children(A), [B])
+
+        # 4) algophony batch query retrieves germ generations from the shared store.
+        germ_generations = [r["akousma_id"] for r in self.store.query(originating_app="germ")]
+        self.assertIn(B, germ_generations)
+        self.assertEqual(len(self.store.query()), 2)  # both A and B live in one store
+
+
+if __name__ == "__main__":
+    unittest.main()