hyperpolymath · hyperpolymath · Apr 21, 2026 · Apr 16, 2026 · Apr 19, 2026 · Apr 19, 2026
@@ -6,7 +6,7 @@
 [metadata]
 project = "burble"
 version = "1.1.0-pre"
-last-updated = "2026-04-16"
+last-updated = "2026-04-21"
 status = "active"
 
 [project-context]
@@ -26,7 +26,7 @@ milestones = [
   { name = "Phase 0 — Scrub baseline (V-lang removed, docs honest)", completion = 100, date = "2026-04-16" },
   { name = "Phase 1 — Audio dependable (Opus honest, comfort noise, REMB, Avow chain, echo-cancel ref, neural spectral-gate verified)", completion = 85 },
   { name = "Phase 2 — P2P AI channel dependable (burble-ai-bridge fixes, round-trip tests, docs) — CRITICAL PATH for family/pair-programming use case", completion = 80 },
-  { name = "Phase 2b — server-side Burble.LLM (provider, circuit breaker, fixed parse_frame, NimblePool wired) — SECONDARY, not required for family use case", completion = 40 },
+  { name = "Phase 2b — server-side Burble.LLM (provider, circuit breaker, fixed parse_frame, NimblePool wired) — SECONDARY, not required for family use case", completion = 80 },
   { name = "Phase 3 — RTSP + signaling + text + AffineScript client start", completion = 0 },
   { name = "Phase 4 — PTP hardware clock via Zig NIF, phc2sys supervisor, multi-node align", completion = 10 },
   { name = "Phase 5 — ReScript -> AffineScript completion", completion = 0 }
@@ -39,7 +39,7 @@ signaling-relay = { status = "consolidated", canonical = "signaling/relay.js", r
 
 [blockers-and-issues]
 doc-reality-drift = [
-  "ROADMAP.adoc claims LLM Service DONE — is a stub (provider missing, parse_frame broken)",
+  "ROADMAP.adoc LLM — AnthropicProvider wired, NimblePool gating, REST endpoint live. Remaining: circuit breaker, rate limiting per user, streaming SSE endpoint.",
   "ROADMAP.adoc claims Formal Proofs DONE — Avow attestation is data-type-only, no dependent-type enforcement",
   "README.adoc PTP claim sub-microsecond assumes hardware — code falls back to system clock without NIF"
 ]
@@ -66,7 +66,7 @@ phase-1-audio = [
   "DONE 2026-04-16: Server-side comfort noise injection after 3 silent frames (60ms at 20ms/frame)",
   "DONE 2026-04-16: REMB bitrate adaptation — Pipeline.update_bitrate/3 wired via Backend.io_adaptive_bitrate",
   "DONE 2026-04-16: Avow hash-chain linkage + ETS store + 10 property tests (commit 43669aa)",
-  "NEXT: Wire RTP-timestamp sync in media/peer.ex → Pipeline (precursor to PTP Phase 4, not blocking audio)"
+  "DONE 2026-04-21: Wire RTP-timestamp sync — peer.ex extracts packet.timestamp, pipeline.ex stores last_rtp_ts via record_rtp_timestamp/2 cast (Phase 4 PTP correlation ready)"
 ]
 
 [maintenance-status]

@@ -23,9 +23,10 @@ Console.log2("[Burble] Auth:", AuthState.displayName(app.auth))
 @val @scope("window")
 external addPopStateListener: (@as("popstate") _, affine ('a => unit)) => unit = "addEventListener"
 
+// The closure borrows app via &app (shared borrow) — does not consume it.
 addPopStateListener(_ => {
   let path: string = %raw(`window.location.pathname`)
-  App.handleUrlChange(app, path)
+  App.handleUrlChange(&app, path)
 })
 
 // Set initial page title (borrows app.currentRoute).

@@ -57,6 +57,9 @@ if config_env() == :prod do
     enabled: true,
     primary_port: String.to_integer(System.get_env("LLM_PORT") || "8503"),
     fallback_port: String.to_integer(System.get_env("LLM_FALLBACK_PORT") || "8085"),
+    # Anthropic Claude API — set ANTHROPIC_API_KEY to enable server-side LLM.
+    anthropic_model: System.get_env("ANTHROPIC_MODEL") || "claude-sonnet-4-6",
+    anthropic_max_tokens: String.to_integer(System.get_env("ANTHROPIC_MAX_TOKENS") || "4096"),
     ipv6_preference: true,
     tls: [
       certfile: "priv/ssl/cert.pem",

@@ -140,6 +140,10 @@ defmodule Burble.Coprocessor.Pipeline do
       # Silence counter: frames since last non-nil inbound. Drives comfort
       # noise injection so peers don't hear dead air when a speaker pauses.
       silence_frames: 0,
+      # Last RTP timestamp received from the network — populated via
+      # record_rtp_timestamp/2 when peer.ex extracts it from incoming packets.
+      # Used by Phase 4 PTP correlation to map RTP clock → wall clock.
+      last_rtp_ts: 0,
       # Metrics
       frames_processed: 0,
       frames_dropped: 0,
@@ -297,6 +301,27 @@ defmodule Burble.Coprocessor.Pipeline do
     {:reply, {:ok, health}, state}
   end
 
+  # ---------------------------------------------------------------------------
+  # RTP timestamp tracking (Phase 4 PTP precursor)
+  # ---------------------------------------------------------------------------
+
+  @doc """
+  Record the latest RTP timestamp received from the network.
+
+  Called by `Burble.Media.Peer` each time an RTP packet arrives so the
+  pipeline knows the sender's RTP clock position. Phase 4 will correlate
+  this against the PTP hardware clock to derive end-to-end latency and
+  enable multi-node playout alignment.
+  """
+  def record_rtp_timestamp(pipeline, rtp_ts) do
+    GenServer.cast(pipeline, {:rtp_timestamp, rtp_ts})
+  end
+
+  @impl true
+  def handle_cast({:rtp_timestamp, rtp_ts}, state) do
+    {:noreply, %{state | last_rtp_ts: rtp_ts}}
+  end
+
   # ---------------------------------------------------------------------------
   # Bitrate adaptation (REMB feedback)
   # ---------------------------------------------------------------------------

@@ -517,10 +517,10 @@ defmodule Burble.Coprocessor.SNIFBackend do
           result = Wasmex.call_function(pid, function, args)
           GenServer.stop(pid, :normal)
           result
-        {:error, reason} -> {:error, :wasm_load_failed, detail: reason}
+        {:error, reason} -> {:error, {:wasm_load_failed, reason}}
       end
     rescue
-      error -> {:error, :snif_exception, detail: error}
+      error -> {:error, {:snif_exception, error}}
     end
   end
 

@@ -21,35 +21,58 @@ defmodule Burble.LLM do
     :ok
   end
 
+  @pool_timeout 65_000
+
   @doc """
-  Process a synchronous LLM query.
+  Process a synchronous LLM query. Routes through the NimblePool to gate
+  concurrency — at most `pool_size` (default 10) requests run simultaneously.
   """
   def process_query(user_id, prompt) do
-    Logger.debug("[LLM] Processing query for #{user_id}: #{prompt}")
+    Logger.debug("[LLM] Processing query for #{user_id}")
     provider = :persistent_term.get({__MODULE__, :provider}, @provider)
 
     if provider do
-      provider.process_query(user_id, prompt)
+      checkout_and_run(fn -> provider.process_query(user_id, prompt) end)
     else
       Logger.warning("[LLM] process_query called but no provider is configured")
       {:error, :no_provider_configured}
     end
   end
 
   @doc """
-  Stream an LLM query response.
+  Stream an LLM query response. Routes through the NimblePool to gate
+  concurrency — the pool slot is held for the duration of the stream.
   """
   def stream_query(user_id, prompt, callback) do
-    Logger.debug("[LLM] Streaming query for #{user_id}: #{prompt}")
+    Logger.debug("[LLM] Streaming query for #{user_id}")
     provider = :persistent_term.get({__MODULE__, :provider}, @provider)
 
     if provider do
-      provider.stream_query(user_id, prompt, callback)
+      checkout_and_run(fn -> provider.stream_query(user_id, prompt, callback) end)
     else
       Logger.warning("[LLM] stream_query called but no provider is configured")
       {:error, :no_provider_configured}
     end
   end
+
+  # Checkout a worker from the pool, run the function, then check back in.
+  # If the pool is exhausted, the caller blocks until a slot opens or timeout.
+  defp checkout_and_run(fun) do
+    try do
+      NimblePool.checkout!(:llm_worker_pool, :checkout, fn _from, worker_state ->
+        result = fun.()
+        {result, worker_state}
+      end, @pool_timeout)
+    catch
+      :exit, {:timeout, _} ->
+        Logger.warning("[LLM] Pool checkout timeout — all workers busy")
+        {:error, :pool_exhausted}
+
+      :exit, reason ->
+        Logger.error("[LLM] Pool checkout failed: #{inspect(reason)}")
+        {:error, :pool_error}
+    end
+  end
 end
 
 defmodule Burble.LLM.Registry do

@@ -0,0 +1,180 @@
+# SPDX-License-Identifier: PMPL-1.0-or-later
+# Copyright (c) 2026 Jonathan D.A. Jewell (hyperpolymath)
+
+defmodule Burble.LLM.AnthropicProvider do
+  @moduledoc """
+  Anthropic Claude API provider for Burble's LLM service.
+
+  Calls the Claude Messages API via Erlang's built-in :httpc (no extra deps).
+  Reads ANTHROPIC_API_KEY from environment. Falls back gracefully when unconfigured.
+  """
+
+  require Logger
+
+  @api_url ~c"https://api.anthropic.com/v1/messages"
+  @api_version "2023-06-01"
+  @default_model "claude-sonnet-4-6"
+  @default_max_tokens 4096
+  @request_timeout 60_000
+
+  # Ensure inets + ssl are started (idempotent).
+  defp ensure_httpc do
+    :inets.start()
+    :ssl.start()
+    :ok
+  end
+
+  @doc """
+  Process a synchronous LLM query. Returns `{:ok, text}` or `{:error, reason}`.
+  """
+  def process_query(user_id, prompt) do
+    ensure_httpc()
+
+    case api_key() do
+      nil ->
+        {:error, :api_key_not_configured}
+
+      key ->
+        body = Jason.encode!(%{
+          model: model(),
+          max_tokens: max_tokens(),
+          messages: [%{role: "user", content: prompt}],
+          system: system_prompt(user_id)
+        })
+
+        headers = [
+          {~c"content-type", ~c"application/json"},
+          {~c"x-api-key", String.to_charlist(key)},
+          {~c"anthropic-version", String.to_charlist(@api_version)}
+        ]
+
+        request = {@api_url, headers, ~c"application/json", String.to_charlist(body)}
+
+        case :httpc.request(:post, request, [timeout: @request_timeout, ssl: ssl_opts()], []) do
+          {:ok, {{_, 200, _}, _resp_headers, resp_body}} ->
+            parse_response(List.to_string(resp_body))
+
+          {:ok, {{_, status, _}, _resp_headers, resp_body}} ->
+            Logger.warning("[LLM/Anthropic] API returned #{status}: #{List.to_string(resp_body)}")
+            {:error, {:api_error, status}}
+
+          {:error, reason} ->
+            Logger.error("[LLM/Anthropic] HTTP request failed: #{inspect(reason)}")
+            {:error, {:http_error, reason}}
+        end
+    end
+  end
+
+  @doc """
+  Stream an LLM query response. Calls `callback.(chunk_text)` for each content delta.
+  Returns `:ok` on completion or `{:error, reason}`.
+  """
+  def stream_query(user_id, prompt, callback) do
+    ensure_httpc()
+
+    case api_key() do
+      nil ->
+        {:error, :api_key_not_configured}
+
+      key ->
+        body = Jason.encode!(%{
+          model: model(),
+          max_tokens: max_tokens(),
+          stream: true,
+          messages: [%{role: "user", content: prompt}],
+          system: system_prompt(user_id)
+        })
+
+        headers = [
+          {~c"content-type", ~c"application/json"},
+          {~c"x-api-key", String.to_charlist(key)},
+          {~c"anthropic-version", String.to_charlist(@api_version)}
+        ]
+
+        request = {@api_url, headers, ~c"application/json", String.to_charlist(body)}
+
+        case :httpc.request(:post, request, [timeout: @request_timeout, ssl: ssl_opts()], []) do
+          {:ok, {{_, 200, _}, _resp_headers, resp_body}} ->
+            parse_sse_stream(List.to_string(resp_body), callback)
+
+          {:ok, {{_, status, _}, _resp_headers, resp_body}} ->
+            Logger.warning("[LLM/Anthropic] Stream API returned #{status}")
+            {:error, {:api_error, status, List.to_string(resp_body)}}
+
+          {:error, reason} ->
+            Logger.error("[LLM/Anthropic] Stream HTTP request failed: #{inspect(reason)}")
+            {:error, {:http_error, reason}}
+        end
+    end
+  end
+
+  # ---------------------------------------------------------------------------
+  # Response parsing
+  # ---------------------------------------------------------------------------
+
+  defp parse_response(body) do
+    case Jason.decode(body) do
+      {:ok, %{"content" => [%{"text" => text} | _]}} ->
+        {:ok, text}
+
+      {:ok, %{"error" => %{"message" => msg}}} ->
+        {:error, {:api_error, msg}}
+
+      {:ok, other} ->
+        Logger.warning("[LLM/Anthropic] Unexpected response shape: #{inspect(other)}")
+        {:error, :unexpected_response}
+
+      {:error, _} ->
+        {:error, :json_decode_error}
+    end
+  end
+
+  defp parse_sse_stream(body, callback) do
+    body
+    |> String.split("\n")
+    |> Enum.each(fn line ->
+      case line do
+        "data: " <> json_str ->
+          case Jason.decode(json_str) do
+            {:ok, %{"type" => "content_block_delta", "delta" => %{"text" => text}}} ->
+              callback.(text)
+            _ ->
+              :ok
+          end
+        _ ->
+          :ok
+      end
+    end)
+
+    :ok
+  end
+
+  # ---------------------------------------------------------------------------
+  # Config helpers
+  # ---------------------------------------------------------------------------
+
+  defp api_key do
+    System.get_env("ANTHROPIC_API_KEY")
+  end
+
+  defp model do
+    System.get_env("ANTHROPIC_MODEL") || @default_model
+  end
+
+  defp max_tokens do
+    case System.get_env("ANTHROPIC_MAX_TOKENS") do
+      nil -> @default_max_tokens
+      val -> String.to_integer(val)
+    end
+  end
+
+  defp system_prompt(user_id) do
+    "You are a helpful AI assistant integrated into Burble, a P2P voice and collaboration platform. " <>
+    "You are responding to user #{user_id}. Be concise, technical when appropriate, and helpful. " <>
+    "If the user asks about Burble features, you can mention voice chat, AI bridge, E2EE, and WebRTC."
+  end
+
+  defp ssl_opts do
+    [verify: :verify_peer, cacerts: :public_key.cacerts_get(), depth: 3]
+  end
+end