@@ -0,0 +1,43 @@
+import os
+import re
+import subprocess
+from pathlib import Path
+
+DEFAULT_PULSE_SOCKET = "/tmp/pulse-socket"
+_PERCENT = re.compile(r"(\d+)%")
+
+
+def _env() -> dict:
+    env = dict(os.environ)
+    if "PULSE_SERVER" not in env and Path(DEFAULT_PULSE_SOCKET).exists():
+        env["PULSE_SERVER"] = f"unix:{DEFAULT_PULSE_SOCKET}"
+    return env
+
+
+def _pactl(*args: str, timeout: float = 5.0) -> subprocess.CompletedProcess:
+    return subprocess.run(["pactl", *args], capture_output=True, text=True, timeout=timeout, env=_env())
+
+
+def get_volume() -> dict | None:
+    try:
+        volume = _pactl("get-sink-volume", "@DEFAULT_SINK@")
+        mute = _pactl("get-sink-mute", "@DEFAULT_SINK@")
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    if volume.returncode != 0:
+        return None
+    levels = [int(m) for m in _PERCENT.findall(volume.stdout)]
+    if not levels:
+        return None
+    return {"percent": max(levels), "muted": mute.returncode == 0 and "yes" in mute.stdout.lower()}
+
+
+def set_volume(percent: int | None = None, muted: bool | None = None) -> dict | None:
+    try:
+        if percent is not None:
+            _pactl("set-sink-volume", "@DEFAULT_SINK@", f"{max(0, min(100, int(percent)))}%")
+        if muted is not None:
+            _pactl("set-sink-mute", "@DEFAULT_SINK@", "1" if muted else "0")
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    return get_volume()
@@ -0,0 +1,78 @@
+import re
+import subprocess
+from pathlib import Path
+
+
+def _run(argv: list[str], timeout: float = 5.0) -> str:
+    try:
+        result = subprocess.run(argv, capture_output=True, text=True, timeout=timeout)
+    except (OSError, subprocess.TimeoutExpired):
+        return ""
+    return result.stdout if result.returncode == 0 else ""
+
+
+def _split_terse(line: str) -> list[str]:
+    parts, current, escaped = [], "", False
+    for ch in line:
+        if escaped:
+            current += ch
+            escaped = False
+        elif ch == "\\":
+            escaped = True
+        elif ch == ":":
+            parts.append(current)
+            current = ""
+        else:
+            current += ch
+    parts.append(current)
+    return parts
+
+
+def wan_route() -> dict:
+    """Interface and address the kernel uses to reach the internet."""
+    out = _run(["ip", "-4", "route", "get", "1.1.1.1"])
+    dev = re.search(r"\bdev (\S+)", out)
+    src = re.search(r"\bsrc (\S+)", out)
+    via = re.search(r"\bvia (\S+)", out)
+    return {
+        "interface": dev.group(1) if dev else None,
+        "ip": src.group(1) if src else None,
+        "gateway": via.group(1) if via else None,
+    }
+
+
+def interface_kind(interface: str | None) -> str | None:
+    if not interface:
+        return None
+    if Path(f"/sys/class/net/{interface}/wireless").exists():
+        return "wifi"
+    return "ethernet"
+
+
+def current_ssid() -> str | None:
+    for line in _run(["nmcli", "-t", "-f", "ACTIVE,SSID", "device", "wifi"]).splitlines():
+        fields = _split_terse(line)
+        if len(fields) >= 2 and fields[0] == "yes":
+            return fields[1] or None
+    return None
+
+
+def interface_addresses() -> list[dict]:
+    rows = []
+    for line in _run(["ip", "-4", "-o", "addr"]).splitlines():
+        match = re.match(r"\d+:\s+(\S+)\s+inet\s+(\S+)", line)
+        if match and match.group(1) != "lo":
+            rows.append({"interface": match.group(1), "ip": match.group(2).split("/")[0]})
+    return rows
+
+
+def network_info() -> dict:
+    wan = wan_route()
+    kind = interface_kind(wan["interface"])
+    return {
+        **wan,
+        "kind": kind,
+        "ssid": current_ssid() if kind == "wifi" else None,
+        "online": wan["interface"] is not None,
+        "addresses": interface_addresses(),
+    }
@@ -2,6 +2,7 @@
 from typing import Callable
 
 from robot_orchestrator.bootlog import logged_run
+from robot_orchestrator.network.info import _split_terse
 from robot_orchestrator.secrets.protocol import WifiCredential
 
 RunnerT = Callable[[list[str], float], subprocess.CompletedProcess]
@@ -31,6 +32,56 @@ def list_visible_ssids(self) -> list[str]:
                 ssids.append(ssid)
         return ssids
 
+    def list_networks(self) -> list[dict]:
+        try:
+            result = self.runner(
+                ["nmcli", "-t", "-f", "IN-USE,SSID,SIGNAL,SECURITY", "device", "wifi", "list"], self.timeout_s
+            )
+        except (OSError, subprocess.TimeoutExpired):
+            return []
+        if result.returncode != 0:
+            return []
+        networks: dict[str, dict] = {}
+        for line in result.stdout.splitlines():
+            fields = _split_terse(line)
+            if len(fields) < 4 or not fields[1]:
+                continue
+            item = {
+                "ssid": fields[1], "signal": int(fields[2]) if fields[2].isdigit() else 0,
+                "secured": fields[3] not in ("", "--"), "in_use": fields[0] == "*",
+            }
+            known = networks.get(item["ssid"])
+            if known is None or (item["in_use"], item["signal"]) > (known["in_use"], known["signal"]):
+                networks[item["ssid"]] = item
+        return sorted(networks.values(), key=lambda n: (not n["in_use"], -n["signal"]))
+
+    def saved_profiles(self) -> list[str]:
+        try:
+            result = self.runner(["nmcli", "-t", "-f", "NAME,TYPE", "connection", "show"], self.timeout_s)
+        except (OSError, subprocess.TimeoutExpired):
+            return []
+        if result.returncode != 0:
+            return []
+        names = []
+        for line in result.stdout.splitlines():
+            fields = _split_terse(line)
+            if len(fields) >= 2 and fields[1] == "802-11-wireless":
+                names.append(fields[0])
+        return names
+
+    def rescan(self) -> None:
+        try:
+            self.runner(["nmcli", "device", "wifi", "rescan"], self.timeout_s)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+
+    def connect_saved(self, ssid: str) -> bool:
+        try:
+            result = self.runner(["nmcli", "connection", "up", "id", ssid], self.timeout_s * 2)
+        except (OSError, subprocess.TimeoutExpired):
+            return False
+        return result.returncode == 0
+
     def connect(self, ssid: str, psk: str) -> bool:
         try:
             result = self.runner(["nmcli", "device", "wifi", "connect", ssid, "password", psk], self.timeout_s)
 
@@ -157,6 +157,17 @@ def boot_state(request: Request):
             "operator_url": context.operator_url,
         }
 
+    @app.post("/boot/api/session", include_in_schema=False)
+    def boot_session(request: Request):
+        require_loopback(request)
+        # a tunnel or reverse proxy ends up as a loopback peer too, but it always adds forwarding headers
+        if any(h in request.headers for h in ("x-forwarded-for", "forwarded", "x-real-ip")):
+            raise HTTPException(status_code=403, detail="direct loopback only")
+        if not (context and context.boot_progress and context.boot_progress.finished_at):
+            raise HTTPException(status_code=409, detail="boot is not finished yet")
+        token = auth.issue_token(jwt_secret, "boot-settings", jwt_ttl_s)
+        return {"access_token": token, "expires_in": jwt_ttl_s}
+
     @app.get("/boot/api/stream", include_in_schema=False)
     async def boot_stream(request: Request):
         require_loopback(request)
 
@@ -5,16 +5,33 @@
 from fastapi import Depends, FastAPI, HTTPException
 from pydantic import BaseModel
 
+from robot_orchestrator.audio import volume as audio_volume
+from robot_orchestrator.network import info as network_info
 from robot_orchestrator.secrets import store as secrets_store
 from robot_orchestrator.secrets.ingest import ingest_payload
 from robot_orchestrator.secrets.protocol import SecretsPayload
 from robot_orchestrator.state import store
 from robot_orchestrator.supervisor.logsink import parse_line
 from robot_orchestrator.supervisor.service import ServiceState
 from robot_orchestrator.web import auth
+from robot_orchestrator.web import voice_control
 from robot_orchestrator.web.context import AdminContext
 
 
+class WifiJoinRequest(BaseModel):
+    ssid: str
+    psk: str | None = None
+
+
+class VolumeRequest(BaseModel):
+    percent: int | None = None
+    muted: bool | None = None
+
+
+class WakeWordRequest(BaseModel):
+    enabled: bool
+
+
 class ChangePasswordRequest(BaseModel):
     current_password: str
     new_password: str
@@ -248,6 +265,64 @@ async def connect_wifi(claims: dict = Depends(require_auth)):
         connected = await asyncio.to_thread(c.wifi_manager.try_known_networks, known)
         return {"connected_ssid": connected}
 
+    @app.get("/api/network")
+    async def get_network(claims: dict = Depends(require_auth)):
+        return await asyncio.to_thread(network_info.network_info)
+
+    @app.get("/api/wifi/networks")
+    async def get_wifi_networks(rescan: bool = False, claims: dict = Depends(require_auth)):
+        c = _require(context)
+        if c.wifi_manager is None:
+            raise HTTPException(status_code=503, detail="wifi manager is not available")
+        if rescan:
+            await asyncio.to_thread(c.wifi_manager.rescan)
+        networks = await asyncio.to_thread(c.wifi_manager.list_networks)
+        saved = await asyncio.to_thread(c.wifi_manager.saved_profiles)
+        return {"networks": [{**n, "saved": n["ssid"] in saved} for n in networks]}
+
+    @app.post("/api/wifi/join")
+    async def join_wifi(body: WifiJoinRequest, claims: dict = Depends(require_auth)):
+        c = _require(context)
+        if c.wifi_manager is None:
+            raise HTTPException(status_code=503, detail="wifi manager is not available")
+        if body.psk:
+            connected = await asyncio.to_thread(c.wifi_manager.connect, body.ssid, body.psk)
+        else:
+            connected = await asyncio.to_thread(c.wifi_manager.connect_saved, body.ssid)
+        return {"connected": connected, "ssid": body.ssid}
+
+    @app.get("/api/audio/volume")
+    async def get_volume(claims: dict = Depends(require_auth)):
+        state = await asyncio.to_thread(audio_volume.get_volume)
+        if state is None:
+            raise HTTPException(status_code=503, detail="audio mixer is not available")
+        return state
+
+    @app.post("/api/audio/volume")
+    async def set_volume(body: VolumeRequest, claims: dict = Depends(require_auth)):
+        state = await asyncio.to_thread(audio_volume.set_volume, body.percent, body.muted)
+        if state is None:
+            raise HTTPException(status_code=503, detail="audio mixer is not available")
+        return state
+
+    @app.get("/api/voice/wake-word")
+    async def get_wake_word(claims: dict = Depends(require_auth)):
+        c = _require(context)
+        try:
+            status = await asyncio.to_thread(voice_control.voice_status, c)
+        except voice_control.VoiceControlError as e:
+            raise HTTPException(status_code=e.status, detail=str(e))
+        return {"connected": bool(status.get("connected")), "enabled": status.get("wake_word_enabled")}
+
+    @app.post("/api/voice/wake-word")
+    async def set_wake_word(body: WakeWordRequest, claims: dict = Depends(require_auth)):
+        c = _require(context)
+        try:
+            status = await asyncio.to_thread(voice_control.set_wake_word, c, body.enabled)
+        except voice_control.VoiceControlError as e:
+            raise HTTPException(status_code=e.status, detail=str(e))
+        return {"connected": bool(status.get("connected")), "enabled": status.get("wake_word_enabled")}
+
     @app.post("/api/settings/password")
     def change_password(body: ChangePasswordRequest, claims: dict = Depends(require_auth)):
         c = _require(context)