Hands-On: Put a TTS Bridge Through a Model-Free Test Bench
In the main Part 9 article, I separated the speech API an agent sees from the engine that generates audio and owns its references. This lab gives you a safe way to exercise that boundary without installing a model, borrowing somebody’s voice, or pointing a client at a network provider.
I built the exercise to prove the operational contract, not to create a toy voice-cloning demo. It generates a short tone WAV and an invented transcript, starts a fake OpenAI-compatible speech engine and a small teaching bridge on ephemeral loopback ports, and walks through both successful requests and the failures I care about. When it finishes, both servers stop and the temporary material disappears with them.
The preferred path uses an opaque registered-reference ID. One synthetic path pair remains so the older compatibility contract is visible and testable. The tone is not speech and does not represent a person. Nothing in the package measures voice similarity, speaker identity, model quality, GPU behavior, or production latency.
Moving from this lab to a real engine means replacing the generated tone with a recording you have the right to use. That can be your own voice, purpose-made synthetic material, or a recording whose speaker gave explicit permission. If the model is transcript-conditioned, you also need to write down exactly what was said and keep that transcript with the recording when you register it. Some supported models use audio alone, so check the contract for the model you chose instead of copying the Qwen requirements blindly. The lab deliberately gives you neither a real voice nor a transcript to reuse.
What You Will Build and Verify
The package contains a teaching bridge, a fake upstream, a complete runner, seven regression tests, a README, and an intentionally empty dependency declaration. Everything runs on the Python standard library.
| Check | Required result |
|---|---|
| Metadata discovery | The bridge caches capability data plus registry reachability and count for a bounded interval |
| Alias source | The bridge exposes its configured aliases and does not retain the registry’s reference IDs |
| Preferred alias | Case-insensitive NARRATOR becomes one opaque reference_id |
| Boundary check | No audio or transcript path crosses the bridge for the registered alias |
| Legacy compatibility | guide selects the generated audio and transcript path pair |
| Validation failures | An unavailable registry refresh returns 502 and an unsupported instruction returns 422, both before synthesis |
| Format compatibility | An OGG request is explicitly delivered as WAV |
| Operational events | Input and reference details are replaced with redaction markers |
| Split health | Bridge health remains 200 after the upstream stops, while synthesis returns 502 |
| Cleanup | Both server threads stop and the temporary directory is removed |
Step 1: Put the Lab in an Isolated Directory
Copy the five companion files into an empty working directory:
README.md
requirements.txt
run_lab.py
test_lab.py
tts_bridge_lab.py
Download the complete five-file Hands-On 9A package. Every file is also available below through the site’s standard collapsed source viewer.
README.md markdown View source
# Hands-On 9A: Model-Free TTS Bridge Lab
This lab demonstrates an OpenAI-compatible TTS compatibility boundary without loading a model or using a real voice. It generates a short tone WAV and an invented transcript in a temporary directory, starts a fake speech upstream with capability and reference-registry endpoints plus a teaching bridge on ephemeral loopback ports, exercises a bounded metadata cache and capability-driven validation, and removes the temporary material during cleanup.
## Safety and Scope
The generated tone is not speech and does not represent a person. The aliases `narrator` and `guide` are invented. The lab uses no network provider, credential, production path, private sample, model, GPU, or external dependency. It tests request and operational behavior, not voice-cloning quality.
## Run
```bash
python3 run_lab.py
python3 -m unittest -v test_lab.py
```
Both commands should exit zero. The runner prints a bounded result table containing status codes and Boolean checks. It never prints input text, audio bytes, temporary paths, or a voice sample.
## What the Lab Proves
The complete run checks that:
- bridge and upstream health are separate;
- a case-insensitive neutral alias becomes an opaque registered-reference ID;
- no audio or transcript path crosses the bridge for that preferred request;
- capability data plus registry reachability and count are cached and reused without retaining registry IDs;
- one synthetic path pair remains as explicit legacy compatibility coverage;
- an unavailable registry fails before synthesis with 502;
- an unsupported instruction control fails before synthesis with 422;
- an OGG request is explicitly delivered as WAV;
- the returned bytes match the generated tone;
- the bridge's operational event contains markers rather than input or reference details;
- invalid input returns 400;
- an upstream timeout returns 502;
- bridge health remains 200 after the upstream has stopped while the next speech request returns 502;
- both servers stop and the temporary reference directory is removed.
## What the Lab Does Not Prove
The lab does not test a TTS model, voice similarity, speaker identity, consent, inference quality, GPU behavior, production latency, remote networking, launchd, or automatic restart. The teaching bridge is deliberately small and is not a drop-in replacement for the current LLM-Ops-Kit bridge.
## Files
| File | Purpose |
| --- | --- |
| `tts_bridge_lab.py` | Fake upstream, teaching bridge, alias resolution, format normalization, generated tone, and managed server helpers |
| `run_lab.py` | Complete acceptance sequence and bounded report |
| `test_lab.py` | Seven regression tests, including the full contract, preferred registered reference, legacy compatibility, redaction leak rejection, and response-header line-break removal |
| `requirements.txt` | Confirms that the lab uses only the Python standard library |
## Cleanup
The runner stops both loopback servers in a `finally` block. Its `TemporaryDirectory` owns the tone and transcript and removes them after the server cleanup completes. If the process is interrupted, no persistent configuration or production service has been modified; the operating system can remove any abandoned temporary directory through its normal temporary-file lifecycle.
requirements.txt text View source
# Standard library only. Tested with Python 3.12.
run_lab.py python View source
#!/usr/bin/env python3
"""Run the complete model-free TTS Bridge lab."""
from __future__ import annotations
import json
import tempfile
from pathlib import Path
from typing import Any
from urllib import error, request
from tts_bridge_lab import (
BridgeConfig,
FakeUpstreamState,
make_bridge,
make_fake_upstream,
make_tone_wav,
operational_events_redact_inputs,
operational_events_redact_references,
)
SYNTHESIS_INPUTS = (
"Invented lab sentence.",
"Invented legacy compatibility sentence.",
"Invented registry failure case.",
"Invented unsupported-control case.",
"Invented timeout case.",
"Invented unavailable-upstream case.",
)
def get_json(url: str) -> tuple[int, dict[str, Any]]:
try:
with request.urlopen(url, timeout=2) as response:
return response.status, json.loads(response.read().decode())
except error.HTTPError as exc:
return exc.code, json.loads(exc.read().decode())
def post_json(url: str, value: dict[str, Any]) -> tuple[int, bytes, dict[str, str]]:
req = request.Request(
url,
data=json.dumps(value).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with request.urlopen(req, timeout=2) as response:
return response.status, response.read(), dict(response.headers.items())
except error.HTTPError as exc:
return exc.code, exc.read(), dict(exc.headers.items())
def execute_lab() -> dict[str, Any]:
with tempfile.TemporaryDirectory(prefix="tts-bridge-lab-") as tmp:
root = Path(tmp)
samples = root / "references"
samples.mkdir()
tone_path = samples / "narrator.wav"
transcript_path = samples / "narrator.txt"
tone_bytes = make_tone_wav(tone_path)
transcript_path.write_text("This is an invented matching transcript.\n", encoding="utf-8")
upstream_state = FakeUpstreamState(audio_bytes=tone_bytes)
upstream = make_fake_upstream(upstream_state)
config = BridgeConfig(
upstream_base=f"http://127.0.0.1:{upstream.port}/v1",
samples_dir=samples,
voice_map={
"narrator": {"reference_id": "ref_lab_narrator"},
"guide": {"sample": tone_path.name, "ref_text": transcript_path.name},
},
timeout_seconds=0.05,
)
bridge = make_bridge(config)
upstream.start()
bridge.start()
bridge_url = f"http://127.0.0.1:{bridge.port}"
upstream_url = f"http://127.0.0.1:{upstream.port}"
try:
bridge_health, _ = get_json(bridge_url + "/health")
upstream_health, _ = get_json(upstream_url + "/health")
speech_status, speech_body, headers = post_json(
bridge_url + "/v1/audio/speech",
{
"input": SYNTHESIS_INPUTS[0],
"voice": "NARRATOR",
"response_format": "ogg",
},
)
registered_forwarded = upstream_state.received[-1]
legacy_status, _, _ = post_json(
bridge_url + "/v1/audio/speech",
{"input": SYNTHESIS_INPUTS[1], "voice": "guide"},
)
legacy_forwarded = upstream_state.received[-1]
metadata_cache_reused = (
upstream_state.capability_requests == 1
and upstream_state.registry_requests == 1
)
invalid_status, _, _ = post_json(
bridge_url + "/v1/audio/speech", {"voice": "narrator"}
)
upstream_state.registry_reachable = False
config.upstream_metadata["checked_at"] = 0.0
count_before_registry_failure = len(upstream_state.received)
registry_unavailable_status, _, _ = post_json(
bridge_url + "/v1/audio/speech",
{"input": SYNTHESIS_INPUTS[2], "voice": "narrator"},
)
registry_failed_before_synthesis = (
len(upstream_state.received) == count_before_registry_failure
)
upstream_state.registry_reachable = True
config.upstream_metadata["checked_at"] = 0.0
count_before_control_failure = len(upstream_state.received)
unsupported_control_status, _, _ = post_json(
bridge_url + "/v1/audio/speech",
{
"input": SYNTHESIS_INPUTS[3],
"voice": "narrator",
"instruction": "Invented unsupported direction.",
},
)
control_failure_precedes_synthesis = (
len(upstream_state.received) == count_before_control_failure
)
upstream_state.delay_seconds = 0.15
timeout_status, _, _ = post_json(
bridge_url + "/v1/audio/speech",
{"input": SYNTHESIS_INPUTS[4], "voice": "narrator"},
)
upstream_state.delay_seconds = 0.0
upstream.stop()
bridge_after_upstream_stop, _ = get_json(bridge_url + "/health")
unavailable_status, _, _ = post_json(
bridge_url + "/v1/audio/speech",
{"input": SYNTHESIS_INPUTS[5], "voice": "narrator"},
)
report = {
"bridge_health": bridge_health,
"upstream_health": upstream_health,
"speech_status": speech_status,
"audio_matches_generated_tone": speech_body == tone_bytes,
"requested_format": headers.get("X-TTS-Bridge-Requested-Format"),
"delivered_format": headers.get("X-TTS-Bridge-Delivered-Format"),
"registered_alias_uses_opaque_id": registered_forwarded
== {
"input": SYNTHESIS_INPUTS[0],
"response_format": "wav",
"reference_id": "ref_lab_narrator",
},
"registered_request_omits_paths": not {
"ref_audio",
"ref_text",
}.intersection(registered_forwarded),
"metadata_cache_reused": metadata_cache_reused,
"legacy_compatibility_status": legacy_status,
"legacy_pair_selected": legacy_forwarded.get("ref_audio")
== str(tone_path.resolve())
and legacy_forwarded.get("ref_text") == str(transcript_path.resolve()),
"registry_unavailable_status": registry_unavailable_status,
"registry_failure_precedes_synthesis": registry_failed_before_synthesis,
"unsupported_control_status": unsupported_control_status,
"control_failure_precedes_synthesis": control_failure_precedes_synthesis,
"input_redacted_in_events": operational_events_redact_inputs(
config.events, SYNTHESIS_INPUTS
),
"references_redacted_in_events": operational_events_redact_references(
config.events,
(
"ref_lab_narrator",
str(tone_path.resolve()),
str(transcript_path.resolve()),
),
),
"invalid_input_status": invalid_status,
"timeout_status": timeout_status,
"bridge_health_after_upstream_stop": bridge_after_upstream_stop,
"request_status_after_upstream_stop": unavailable_status,
}
finally:
if upstream.thread.is_alive():
upstream.stop()
bridge.stop()
report["temporary_reference_directory_removed"] = not root.exists()
report["servers_stopped"] = not upstream.thread.is_alive() and not bridge.thread.is_alive()
return report
def main() -> int:
report = execute_lab()
print("Model-free TTS Bridge lab")
print("=" * 58)
for key, value in report.items():
print(f"{key:42} {value}")
required = {
"bridge_health": 200,
"upstream_health": 200,
"speech_status": 200,
"audio_matches_generated_tone": True,
"requested_format": "ogg",
"delivered_format": "wav",
"registered_alias_uses_opaque_id": True,
"registered_request_omits_paths": True,
"metadata_cache_reused": True,
"legacy_compatibility_status": 200,
"legacy_pair_selected": True,
"registry_unavailable_status": 502,
"registry_failure_precedes_synthesis": True,
"unsupported_control_status": 422,
"control_failure_precedes_synthesis": True,
"input_redacted_in_events": True,
"references_redacted_in_events": True,
"invalid_input_status": 400,
"timeout_status": 502,
"bridge_health_after_upstream_stop": 200,
"request_status_after_upstream_stop": 502,
"temporary_reference_directory_removed": True,
"servers_stopped": True,
}
return 0 if all(report.get(key) == expected for key, expected in required.items()) else 1
if __name__ == "__main__":
raise SystemExit(main())
test_lab.py python View source
#!/usr/bin/env python3
"""Regression tests for the model-free TTS Bridge lab."""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from run_lab import SYNTHESIS_INPUTS, execute_lab
from tts_bridge_lab import (
REDACTION_MARKER,
REFERENCE_REDACTION_MARKER,
BridgeConfig,
build_upstream_payload,
make_tone_wav,
operational_events_redact_inputs,
operational_events_redact_references,
safe_header_value,
)
class TTSBridgeLabTests(unittest.TestCase):
def test_registered_alias_forwards_only_an_opaque_reference_id(self) -> None:
config = BridgeConfig(
upstream_base="http://127.0.0.1:1/v1",
samples_dir=Path("/invented/not-used"),
voice_map={"narrator": {"reference_id": "ref_lab_narrator"}},
)
outgoing, _, _ = build_upstream_payload(
{"input": "Invented text.", "voice": "NARRATOR"}, config
)
self.assertEqual(outgoing["reference_id"], "ref_lab_narrator")
self.assertNotIn("voice", outgoing)
self.assertNotIn("ref_audio", outgoing)
self.assertNotIn("ref_text", outgoing)
def test_alias_selects_audio_and_matching_transcript(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
make_tone_wav(root / "narrator.wav")
(root / "narrator.txt").write_text("Invented transcript.\n", encoding="utf-8")
config = BridgeConfig(
upstream_base="http://127.0.0.1:1/v1",
samples_dir=root,
voice_map={"narrator": {"sample": "narrator.wav"}},
)
outgoing, response_format, fallback = build_upstream_payload(
{"input": "Invented text.", "voice": "NARRATOR", "response_format": "ogg"},
config,
)
self.assertNotIn("voice", outgoing)
self.assertEqual(outgoing["ref_audio"], str((root / "narrator.wav").resolve()))
self.assertEqual(outgoing["ref_text"], str((root / "narrator.txt").resolve()))
self.assertEqual(response_format, "wav")
self.assertEqual(fallback, "ogg")
def test_unknown_alias_remains_an_upstream_voice(self) -> None:
config = BridgeConfig(
upstream_base="http://127.0.0.1:1/v1",
samples_dir=Path("/invented/not-used"),
voice_map={},
)
outgoing, _, _ = build_upstream_payload(
{"input": "Invented text.", "voice": "upstream-default"}, config
)
self.assertEqual(outgoing["voice"], "upstream-default")
self.assertNotIn("ref_audio", outgoing)
self.assertNotIn("ref_text", outgoing)
def test_operational_event_redaction_rejects_every_synthesis_input(self) -> None:
safe_event = f'upstream payload: {{"input": "{REDACTION_MARKER}"}}'
self.assertTrue(operational_events_redact_inputs([safe_event], SYNTHESIS_INPUTS))
for leaked_input in SYNTHESIS_INPUTS:
with self.subTest(leaked_input=leaked_input):
self.assertFalse(
operational_events_redact_inputs(
[safe_event, f"leaked input: {leaked_input}"], SYNTHESIS_INPUTS
)
)
def test_operational_event_redaction_rejects_reference_details(self) -> None:
safe_event = (
'upstream payload: {"input": "<redacted input text>", '
f'"reference_id": "{REFERENCE_REDACTION_MARKER}"}}'
)
sensitive = ("ref_lab_narrator", "/invented/reference.wav")
self.assertTrue(operational_events_redact_references([safe_event], sensitive))
for leaked_reference in sensitive:
with self.subTest(leaked_reference=leaked_reference):
self.assertFalse(
operational_events_redact_references(
[safe_event, f"leaked reference: {leaked_reference}"], sensitive
)
)
def test_header_values_remove_response_splitting_line_breaks(self) -> None:
self.assertEqual(safe_header_value("ogg\r\nX-Injected: yes"), "oggX-Injected: yes")
def test_complete_lab_contract(self) -> None:
report = execute_lab()
self.assertEqual(report["bridge_health"], 200)
self.assertEqual(report["upstream_health"], 200)
self.assertEqual(report["speech_status"], 200)
self.assertTrue(report["audio_matches_generated_tone"])
self.assertEqual(report["requested_format"], "ogg")
self.assertEqual(report["delivered_format"], "wav")
self.assertTrue(report["registered_alias_uses_opaque_id"])
self.assertTrue(report["registered_request_omits_paths"])
self.assertTrue(report["metadata_cache_reused"])
self.assertEqual(report["legacy_compatibility_status"], 200)
self.assertTrue(report["legacy_pair_selected"])
self.assertEqual(report["registry_unavailable_status"], 502)
self.assertTrue(report["registry_failure_precedes_synthesis"])
self.assertEqual(report["unsupported_control_status"], 422)
self.assertTrue(report["control_failure_precedes_synthesis"])
self.assertTrue(report["input_redacted_in_events"])
self.assertTrue(report["references_redacted_in_events"])
self.assertEqual(report["invalid_input_status"], 400)
self.assertEqual(report["timeout_status"], 502)
self.assertEqual(report["bridge_health_after_upstream_stop"], 200)
self.assertEqual(report["request_status_after_upstream_stop"], 502)
self.assertTrue(report["temporary_reference_directory_removed"])
self.assertTrue(report["servers_stopped"])
if __name__ == "__main__":
unittest.main()
tts_bridge_lab.py python View source
#!/usr/bin/env python3
"""Model-free OpenAI-compatible TTS bridge teaching fixture."""
from __future__ import annotations
import io
import json
import math
import struct
import threading
import time
import wave
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib import error, request
FORMAT_FALLBACKS = {"ogg": "wav", "opus": "wav"}
REDACTION_MARKER = "<redacted input text>"
REFERENCE_REDACTION_MARKER = "<redacted reference>"
METADATA_CACHE_SECONDS = 30.0
def safe_header_value(value: str) -> str:
"""Remove line breaks before writing a value to an HTTP response header."""
return value.replace("\r", "").replace("\n", "")
def operational_events_redact_inputs(events: list[str], synthesis_inputs: tuple[str, ...]) -> bool:
event_text = "\n".join(events)
return REDACTION_MARKER in event_text and all(value not in event_text for value in synthesis_inputs)
def operational_events_redact_references(events: list[str], references: tuple[str, ...]) -> bool:
event_text = "\n".join(events)
return REFERENCE_REDACTION_MARKER in event_text and all(
value not in event_text for value in references
)
def make_tone_wav(path: Path, *, seconds: float = 0.08, frequency: float = 440.0) -> bytes:
sample_rate = 16_000
frame_count = int(sample_rate * seconds)
output = io.BytesIO()
with wave.open(output, "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(sample_rate)
for frame in range(frame_count):
value = int(2_000 * math.sin(2.0 * math.pi * frequency * frame / sample_rate))
wav_file.writeframesraw(struct.pack("<h", value))
payload = output.getvalue()
path.write_bytes(payload)
return payload
@dataclass
class FakeUpstreamState:
audio_bytes: bytes
received: list[dict[str, Any]] = field(default_factory=list)
delay_seconds: float = 0.0
fail_status: int | None = None
registry_reachable: bool = True
capability_revision: str = "lab-capability-v1"
capability_requests: int = 0
registry_requests: int = 0
references: list[dict[str, Any]] = field(
default_factory=lambda: [
{
"reference_id": "ref_lab_narrator",
"name": "narrator",
"clone_capable": True,
}
]
)
@dataclass
class BridgeConfig:
upstream_base: str
samples_dir: Path
voice_map: dict[str, dict[str, str]]
timeout_seconds: float = 1.0
events: list[str] = field(default_factory=list)
upstream_metadata: dict[str, Any] = field(default_factory=dict)
class ManagedServer:
def __init__(self, server: ThreadingHTTPServer):
self.server = server
self.thread = threading.Thread(target=server.serve_forever, daemon=True)
def start(self) -> None:
self.thread.start()
@property
def port(self) -> int:
return int(self.server.server_address[1])
def stop(self) -> None:
self.server.shutdown()
self.thread.join(timeout=2)
self.server.server_close()
def normalize_format(requested: str) -> tuple[str, str | None]:
requested = (requested or "wav").lower()
delivered = FORMAT_FALLBACKS.get(requested, requested)
return delivered, requested if delivered != requested else None
def resolve_alias(voice: str, cfg: BridgeConfig) -> dict[str, str] | None:
entry = cfg.voice_map.get((voice or "").strip().lower())
if entry is None:
return None
if "reference_id" in entry:
return {"reference_id": entry["reference_id"]}
sample = (cfg.samples_dir / entry["sample"]).resolve()
transcript_name = entry.get("ref_text") or sample.with_suffix(".txt").name
transcript = (cfg.samples_dir / transcript_name).resolve()
return {"ref_audio": str(sample), "ref_text": str(transcript)}
class DiscoveryUnavailable(RuntimeError):
pass
class CapabilityValidation(ValueError):
pass
def fetch_json(url: str, *, timeout: float) -> dict[str, Any]:
try:
with request.urlopen(url, timeout=timeout) as response:
value = json.loads(response.read().decode("utf-8"))
except Exception as exc: # noqa: BLE001
raise DiscoveryUnavailable("upstream discovery unavailable") from exc
if not isinstance(value, dict):
raise DiscoveryUnavailable("upstream discovery returned an invalid document")
return value
def refresh_upstream_metadata(cfg: BridgeConfig, *, force: bool = False) -> dict[str, Any]:
cached = cfg.upstream_metadata
now = time.monotonic()
if (
not force
and cached
and now - float(cached.get("checked_at", 0.0)) < METADATA_CACHE_SECONDS
):
return cached
base = cfg.upstream_base.rstrip("/")
try:
capabilities = fetch_json(base + "/audio/capabilities", timeout=cfg.timeout_seconds)
registry = fetch_json(base + "/audio/references", timeout=cfg.timeout_seconds)
records = registry.get("data")
if not isinstance(records, list):
raise DiscoveryUnavailable("reference registry returned an invalid document")
refreshed = {
"reachable": True,
"capabilities": capabilities,
"reference_count": len(records),
"checked_at": now,
}
except DiscoveryUnavailable:
refreshed = {
"reachable": False,
"capabilities": {},
"reference_count": 0,
"checked_at": now,
}
cfg.upstream_metadata = refreshed
return refreshed
def apply_style_controls(
output: dict[str, Any], incoming: dict[str, Any], capabilities: dict[str, Any]
) -> None:
instruction = incoming.get("instruction")
if instruction is None:
return
if not isinstance(instruction, str) or not instruction.strip():
raise CapabilityValidation("instruction must be a non-empty string")
clone_active = any(key in output for key in ("ref_audio", "reference_id"))
if clone_active and capabilities.get("family") == "qwen3_tts":
raise CapabilityValidation(
"Qwen reference cloning does not support explicit instruction controls"
)
if "instruct" not in set(capabilities.get("style_controls", [])):
raise CapabilityValidation("loaded model does not support instruction")
output["instruct"] = instruction
def build_upstream_payload(
incoming: dict[str, Any], cfg: BridgeConfig, capabilities: dict[str, Any] | None = None
) -> tuple[dict[str, Any], str, str | None]:
text = incoming.get("input")
if not isinstance(text, str):
raise ValueError("field 'input' must be a string")
delivered_format, downgraded_from = normalize_format(
str(incoming.get("response_format", "wav"))
)
output: dict[str, Any] = {
"input": text,
"response_format": delivered_format,
}
voice = str(incoming.get("voice", ""))
reference = resolve_alias(voice, cfg)
if reference is None:
if voice:
output["voice"] = voice
else:
output.update(reference)
apply_style_controls(output, incoming, capabilities or {})
return output, delivered_format, downgraded_from
class FakeUpstreamHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802
if self.path in ("/health", "/v1/health"):
payload = json.dumps({"ok": True, "kind": "fake-upstream"}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
return
state: FakeUpstreamState = self.server.state # type: ignore[attr-defined]
if self.path == "/v1/audio/capabilities":
state.capability_requests += 1
payload = json.dumps(
{
"revision": state.capability_revision,
"family": "qwen3_tts",
"style_controls": ["instruct"],
}
).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
return
if self.path == "/v1/audio/references":
state.registry_requests += 1
if not state.registry_reachable:
payload = json.dumps({"error": "registry_unavailable"}).encode()
self.send_response(503)
else:
payload = json.dumps({"data": state.references}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
return
self.send_response(404)
self.end_headers()
def do_POST(self) -> None: # noqa: N802
if self.path not in ("/audio/speech", "/v1/audio/speech"):
self.send_response(404)
self.end_headers()
return
state: FakeUpstreamState = self.server.state # type: ignore[attr-defined]
length = int(self.headers.get("Content-Length", "0"))
incoming = json.loads(self.rfile.read(length).decode("utf-8"))
state.received.append(incoming)
if state.delay_seconds:
time.sleep(state.delay_seconds)
if state.fail_status is not None:
payload = json.dumps({"error": "invented upstream failure"}).encode()
self.send_response(state.fail_status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
return
self.send_response(200)
self.send_header("Content-Type", "audio/wav")
self.send_header("Content-Length", str(len(state.audio_bytes)))
self.end_headers()
self.wfile.write(state.audio_bytes)
def log_message(self, fmt: str, *args: object) -> None:
return
class BridgeHandler(BaseHTTPRequestHandler):
def _json(self, status: int, value: dict[str, Any]) -> None:
payload = json.dumps(value).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def do_GET(self) -> None: # noqa: N802
if self.path in ("/health", "/v1/health"):
cfg: BridgeConfig = self.server.config # type: ignore[attr-defined]
metadata = refresh_upstream_metadata(cfg, force=True)
self._json(
200,
{
"ok": True,
"kind": "teaching-bridge",
"upstream": cfg.upstream_base,
"voice_alias_count": len(cfg.voice_map),
"registry_reachable": bool(metadata.get("reachable", False)),
"reference_count": int(metadata.get("reference_count", 0)),
"capability_revision": metadata.get("capabilities", {}).get(
"revision", ""
),
},
)
return
self._json(404, {"error": "not_found"})
def do_POST(self) -> None: # noqa: N802
if self.path not in ("/audio/speech", "/v1/audio/speech"):
self._json(404, {"error": "not_found"})
return
cfg: BridgeConfig = self.server.config # type: ignore[attr-defined]
length = int(self.headers.get("Content-Length", "0"))
try:
incoming = json.loads(self.rfile.read(length).decode("utf-8"))
if not isinstance(incoming, dict):
raise ValueError("payload must be an object")
except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as exc:
self._json(400, {"error": f"input_validation: {exc}"})
return
metadata = refresh_upstream_metadata(cfg)
if not metadata.get("reachable", False):
self._json(502, {"error": "upstream_discovery_unavailable"})
return
try:
outgoing, delivered_format, downgraded_from = build_upstream_payload(
incoming, cfg, metadata.get("capabilities", {})
)
except CapabilityValidation as exc:
self._json(422, {"error": f"style_validation: {exc}"})
return
except ValueError as exc:
self._json(400, {"error": f"input_validation: {exc}"})
return
redacted = dict(outgoing)
redacted["input"] = REDACTION_MARKER
for field in ("reference_id", "ref_audio", "ref_text"):
if field in redacted:
redacted[field] = REFERENCE_REDACTION_MARKER
cfg.events.append("upstream payload: " + json.dumps(redacted, sort_keys=True))
body = json.dumps(outgoing).encode()
req = request.Request(
cfg.upstream_base.rstrip("/") + "/audio/speech",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with request.urlopen(req, timeout=cfg.timeout_seconds) as response:
response_body = response.read()
self.send_response(response.status)
self.send_header("Content-Type", "audio/wav")
self.send_header("Content-Length", str(len(response_body)))
if downgraded_from:
self.send_header(
"X-TTS-Bridge-Requested-Format",
safe_header_value(downgraded_from),
)
self.send_header(
"X-TTS-Bridge-Delivered-Format",
safe_header_value(delivered_format),
)
self.end_headers()
self.wfile.write(response_body)
except error.HTTPError as exc:
self._json(exc.code, {"error": f"upstream_request: HTTP {exc.code}"})
except Exception as exc: # noqa: BLE001
cfg.events.append("upstream failure: " + type(exc).__name__)
self._json(502, {"error": "upstream_request: transport failure"})
def log_message(self, fmt: str, *args: object) -> None:
return
def make_fake_upstream(state: FakeUpstreamState) -> ManagedServer:
server = ThreadingHTTPServer(("127.0.0.1", 0), FakeUpstreamHandler)
server.state = state # type: ignore[attr-defined]
return ManagedServer(server)
def make_bridge(config: BridgeConfig) -> ManagedServer:
server = ThreadingHTTPServer(("127.0.0.1", 0), BridgeHandler)
server.config = config # type: ignore[attr-defined]
return ManagedServer(server)
There is nothing to install. Check that your Python is recent enough and confirm that the dependency file contains no package requirements:
python3 --version
sed -n '1,20p' requirements.txt
The code uses standard-library modules such as http.server, urllib, tempfile, wave, and unittest.
Step 2: Read the Safety Boundary
The runner creates everything under a new TemporaryDirectory. The generated WAV is a low-amplitude tone lasting a fraction of a second. Its transcript says only that it is invented. Both services bind to 127.0.0.1 on ports selected by the operating system.
The lab does not read your TTS configuration, scan a samples directory, inherit a provider token, or contact a remote service. It also does not use launchd or LLM-Ops-Kit. That keeps the exercise focused on the request and observation boundary.
Step 3: Run the Complete Acceptance Sequence
Run the bounded report first:
python3 run_lab.py
The command should exit zero. I deliberately keep exact temporary paths, ephemeral ports, the reference ID, and invented inputs out of the 23-condition report. The useful portion looks like this:
bridge_health 200
upstream_health 200
speech_status 200
audio_matches_generated_tone True
requested_format ogg
delivered_format wav
registered_alias_uses_opaque_id True
registered_request_omits_paths True
metadata_cache_reused True
legacy_compatibility_status 200
legacy_pair_selected True
registry_unavailable_status 502
registry_failure_precedes_synthesis True
unsupported_control_status 422
control_failure_precedes_synthesis True
input_redacted_in_events True
references_redacted_in_events True
invalid_input_status 400
timeout_status 502
bridge_health_after_upstream_stop 200
request_status_after_upstream_stop 502
temporary_reference_directory_removed True
servers_stopped True
This is a deterministic contract report, not a benchmark. It contains status codes, selected-format names, and Boolean comparisons. It does not print audio bytes, request text, reference details, temporary paths, or event payloads.
Step 4: Follow the Preferred Registered Alias
The teaching bridge receives an in-memory alias map with two deliberately different contracts:
voice_map = {
"narrator": {"reference_id": "ref_lab_narrator"},
"guide": {"sample": "narrator.wav", "ref_text": "narrator.txt"},
}
The first entry is the preferred form. The configured alias supplies the invented identifier. The fake registry returns a record, but the teaching bridge retains only registry reachability and count rather than copying its IDs. When the request uses NARRATOR, the bridge removes the alias and forwards only the configured opaque identifier with the target request. The fake engine, like the production engine, remains authoritative for whether that ID is valid.
The runner calls bridge health first, which forces a metadata refresh. The next two synthesis requests reuse the resulting capability and registry metadata instead of performing another discovery round trip. The production cache is bounded to 30 seconds; the lab verifies reuse without sleeping for that interval.
The identifier is visible in the source because it is a synthetic fixture. The product redacts reference content and paths; an opaque ID may remain for operational correlation, but it must stay out of public evidence. I made the lab stricter and redacted the ID from its events too. The string itself is unimportant. What matters is that neither the client nor the bridge needs to know the engine’s filesystem layout.
Step 5: Keep One Legacy Path Pair Honest
The guide alias exercises the older compatibility path. It resolves the generated WAV and matching transcript, then forwards both synthetic paths to the fake upstream. That case is intentionally labeled legacy compatibility rather than presented as the normal cross-host design.
I kept this case because existing deployments sometimes need a transition period, and removing the test would make it easy to break an explicitly supported compatibility path by accident. In a real engine, server paths must be allowlisted, and they should never become the default interface between hosts.
Step 6: Fail Before Synthesis
The runner expires the teaching cache, makes the fake registry return an error, records how many synthesis requests the upstream has received, and calls the bridge again. The failed refresh marks discovery unreachable, the bridge returns 502, and the synthesis count does not change.
It then restores discovery and asks a Qwen-family clone request to use an explicit instruction control. The discovered capability data says that this control is incompatible with reference cloning, so the bridge returns 422 before synthesis. A changed capability revision by itself is not rejected, and the bridge does not validate reference-ID membership locally. Those decisions mirror the current production boundary rather than inventing a stricter bridge contract.
These are not cosmetic checks. They show that an unavailable discovery refresh and a capability-driven control conflict both stop before synthesis, while ID validity remains the engine’s job.
Step 7: Watch Compatibility Without Hiding It
The successful request asks for OGG. The teaching bridge knows that its fake upstream returns WAV, so it sends response_format: wav upstream and returns two headers:
X-TTS-Bridge-Requested-Format: ogg
X-TTS-Bridge-Delivered-Format: wav
The lab compares the response body with the generated tone byte for byte. That proves that the bridge returned the fake upstream’s media unchanged after translating the request. It says nothing about a real codec, model, or player.
Step 8: Separate Health from Request Success
The fake upstream first delays synthesis longer than the bridge’s configured 0.05 second bound. The bridge converts that transport failure into HTTP 502. The short value keeps the lab quick; it is not a production timeout recommendation.
The runner then stops the upstream. The bridge continues to answer its own health endpoint with HTTP 200 because its process and listener are healthy, but that forced health refresh marks discovery unreachable. The next speech request returns HTTP 502 because the required upstream metadata is no longer reachable.
That result is the reason I built the lab. A green bridge does not mean the synthesis path is green. You need separate observations for the bridge, discovery, engine, and a real request.
Step 9: Check Both Redaction Boundaries
Before forwarding, the teaching bridge copies its structured payload for an operational event. It replaces the target input and any reference_id, ref_audio, or ref_text value with markers. The upstream still receives the invented input and required reference because it needs them to fulfill the request.
The runner verifies that none of its six invented inputs, the synthetic reference ID, or either temporary path appears in any event. This is a narrow logging check. It does not prove that every client, reverse proxy, dependency, or crash reporter has the same policy.
Step 10: Run the Regression Tests
Run all seven tests with verbose names:
python3 -m unittest -v test_lab.py
The focused tests cover registered-ID forwarding, legacy pair selection, unknown upstream voices, input redaction, reference redaction, response-header line-break removal, and the complete acceptance sequence. You can also compile every Python file without running the servers:
python3 -m py_compile tts_bridge_lab.py run_lab.py test_lab.py
Step 11: Map the Lab to the Managed Service
The lab deliberately keeps process supervision out of the fixture, but a production investigation cannot stop at a successful request from an interactive shell. In the deployment behind this article, LLM-Ops-Kit manages the bridge as a standalone background component through its dedicated tts-bridge adapter. It is not a launchd job. The bridge depends on the TTS engine, runs from the toolkit’s immutable Python runtime, and uses restart_policy=never, so an intentional stop remains stopped.
Inspect the managed component through the control plane rather than inferring its configuration from a shell environment. Substitute your own qualified component ID:
llmops component status demo-speech:tts-bridge
llmops config effective component demo-speech:tts-bridge
llmops component logs demo-speech:tts-bridge --list
llmops component logs demo-speech:tts-bridge --channel service --lines 200
The effective configuration should make the ownership boundary visible: the bridge belongs on the inference host beside the engine, listens on its client-facing address, and reaches the engine over loopback. The selected configuration revision, component host, execution user, dependency, interpreter, listener, upstream, health target, and log channel should agree. Do not treat an old mutable tree on a non-authority host as current configuration; inspect the authority-selected effective revision.
Then test the protocol in layers. A bridge may legitimately return 404 for an API route it does not implement, so use its actual contract rather than a generic model-server probe:
GET /health
GET /v1/audio/voices
POST /v1/audio/speech
Run those checks first from an authorized operator context, then repeat the speech request through the real client process. Here, Dashboard means the Hermes Agent Dashboard, not a generic web dashboard. That second check matters on macOS because NECP can deny Local Network access to a Dashboard running as a LaunchAgent even when the same URL works under curl in a terminal. NECP is the macOS network-policy subsystem involved in enforcing Local Network privacy decisions. In that failure mode, the client sees No route to host, the bridge records no matching request, and macOS unified logs point back to the client process. Other LAN services may still work because they run in different process and privacy contexts.
The repair is not to move the bridge back or weaken its API. Correct the client process’s Local Network and applicable network-filter permission, or move that client into a reviewed lifecycle context that has the required access. After changing supervision, recheck the component driver, process owner, logs, health, voice discovery, and one real synthesis request. A stack restart is not acceptance by itself: verify every required component returned to running/healthy, and restart a missed client or tunnel explicitly while treating the incomplete stack operation as a control-plane defect to fix.
Cleanup and End State
You should not have to clean up after a successful run. run_lab.py stops the fake upstream and bridge in a finally block, then the temporary directory removes the generated tone and transcript. The final two report lines verify both conditions instead of asking you to trust that cleanup happened.
If you interrupt the process, it still has no authority to modify a production service or configuration. An abandoned temporary directory contains only a generated tone and invented transcript and can be removed through the operating system’s normal temporary-file lifecycle.
What This Lab Cannot Prove
The lab does not load a TTS model, clone a voice, evaluate similarity, measure inference speed, exercise a GPU, test a remote network, validate launchd or another process supervisor, or prove automatic recovery. The operational mapping above is a production diagnostic checklist, not evidence produced by the model-free fixture. The lab does not enforce ownership or consent, and it does not establish that a production bridge supports every third-party provider.
I left those things out on purpose. The useful result is a small executable model of the preferred registered-reference boundary, its legacy compatibility path, and its discovery failures, all without requiring sensitive or expensive material.
Current State
The five-file companion runs with the Python standard library. Its complete 23-condition acceptance runner exits zero, all seven regression tests pass, and all Python files compile. The generated media and transcript are temporary, synthetic, and removed during cleanup. No model, GPU, credential, remote provider, real voice, or private path is used.
Next Work
The next technical step belongs in product acceptance, not this teaching fixture. That means repeating the bridge and TTS protocol checks against the final artifact from both the operator context and the actual client runtime, exercising cold start and outage behavior, confirming stack operations restore every managed process, performing human listening review separately, and testing rollback with authorized material kept inside the inference service boundary.
The proposed restart policy from Part 9 is not part of this lab. It needs its own implementation, desired-state contract, bounded retry tests, explicit-stop test, discovery gating, restart counters, and exhausted-budget evidence before a future Hands-On exercise can present it as runnable behavior.
Join the Discussion
Comments for this post live in GitHub Discussions. That keeps moderation in one place and gives the conversation a stable home.