DecartAI · AdirAmsalem · Jun 18, 2026
diff --git a/src/detail/signaling_channel.cpp b/src/detail/signaling_channel.cpp
@@ -30,7 +30,8 @@ void SignalingChannel::setCallbacks(Callbacks callbacks) { callbacks_ = std::mov
 
 bool SignalingChannel::isOpen() const { return open_; }
 
-RoomInfo SignalingChannel::openAndJoin(std::chrono::milliseconds connectTimeout) {
+RoomInfo SignalingChannel::openAndJoin(std::chrono::milliseconds connectTimeout,
+                                       std::optional<InitialStateAckRequest> initialState) {
   // Single deadline shared across socket-open and room-join phases.
   const auto deadline = std::chrono::steady_clock::now() + connectTimeout;
   const std::string finalUrl = appendQuery(url_, "user_agent", userAgent_);
@@ -80,14 +81,25 @@ RoomInfo SignalingChannel::openAndJoin(std::chrono::milliseconds connectTimeout)
     // fresh connectTimeout so queue waiting does not count against the bound.
     handshakeWindow_ = connectTimeout;
     handshakeDeadline_ = deadline;
+    if (initialState.has_value()) {
+      initialAck_ =
+          InitialAckWatch{std::move(initialState->matchAck), (std::chrono::steady_clock::time_point::max)(),
+                          std::move(initialState->label)};
+    }
   }
-  if (!sendText(makeJoinMsg(std::nullopt).dump())) {
+  const std::optional<nlohmann::json> initialMessage =
+      initialState.has_value() ? std::optional<nlohmann::json>(std::move(initialState->message))
+                               : std::nullopt;
+  if (!sendText(makeJoinMsg(initialMessage).dump())) {
     std::lock_guard<std::mutex> lk(mutex_);
     waiters_.erase(std::remove(waiters_.begin(), waiters_.end(), waiter), waiters_.end());
+    initialAck_.reset();
     throw Exception(Error{ErrorCode::WebsocketError, "Failed to send join"});
   }
 
   // 3. Wait for room_info; queue_position pushes the deadline out.
+  RoomInfo roomInfo;
+  bool shouldStartInitialAckTimer = false;
   {
     std::unique_lock<std::mutex> lk(mutex_);
     while (!waiter->ready && !fatalError_) {
@@ -97,12 +109,24 @@ RoomInfo SignalingChannel::openAndJoin(std::chrono::milliseconds connectTimeout)
       }
     }
     waiters_.erase(std::remove(waiters_.begin(), waiters_.end(), waiter), waiters_.end());
-    if (fatalError_) throw Exception(Error{ErrorCode::ServerError, *fatalError_});
-    if (!waiter->ready) throw Exception(Error{ErrorCode::TimeoutError, "livekit_room_info timeout"});
+    if (fatalError_) {
+      initialAck_.reset();
+      throw Exception(Error{ErrorCode::ServerError, *fatalError_});
+    }
+    if (!waiter->ready) {
+      initialAck_.reset();
+      throw Exception(Error{ErrorCode::TimeoutError, "livekit_room_info timeout"});
+    }
     connected_ = true;
-    return RoomInfo{waiter->result.livekitUrl, waiter->result.token, waiter->result.roomName,
-                    waiter->result.sessionId};
+    if (initialAck_.has_value()) {
+      initialAck_->deadline = std::chrono::steady_clock::now() + initialState->timeout;
+      shouldStartInitialAckTimer = true;
+    }
+    roomInfo = RoomInfo{waiter->result.livekitUrl, waiter->result.token, waiter->result.roomName,
+                        waiter->result.sessionId};
   }
+  if (shouldStartInitialAckTimer) startInitialAckTimer();
+  return roomInfo;
 }
 
 IncomingMessage SignalingChannel::request(const nlohmann::json& message,
@@ -111,6 +135,23 @@ IncomingMessage SignalingChannel::request(const nlohmann::json& message,
   // One ack-bearing request in flight at a time (see requestMutex_).
   std::lock_guard<std::mutex> serialize(requestMutex_);
 
+  std::optional<std::string> initialStateError;
+  {
+    std::unique_lock<std::mutex> lk(mutex_);
+    while (initialAck_.has_value() && !fatalError_) {
+      if (cv_.wait_until(lk, initialAck_->deadline) == std::cv_status::timeout &&
+          std::chrono::steady_clock::now() >= initialAck_->deadline) {
+        initialStateError = initialAck_->label + " timed out";
+        initialAck_.reset();
+        cv_.notify_all();
+        break;
+      }
+    }
+  }
+  if (initialStateError.has_value() && callbacks_.onInitialStateError) {
+    callbacks_.onInitialStateError(*initialStateError);
+  }
+
   auto waiter = std::make_shared<Waiter>();
   waiter->match = std::move(match);
   {
@@ -168,6 +209,29 @@ void SignalingChannel::onMessage(const std::string& text) {
   const IncomingMessage m = parseIncoming(text);
   if (m.type == IncomingType::Unknown) return;
 
+  if (m.type == IncomingType::PromptAck || m.type == IncomingType::SetImageAck) {
+    std::optional<std::string> initialStateError;
+    bool handledInitialAck = false;
+    {
+      std::lock_guard<std::mutex> lk(mutex_);
+      if (initialAck_.has_value() && initialAck_->match(m)) {
+        const std::string label = initialAck_->label;
+        initialAck_.reset();
+        cv_.notify_all();
+        handledInitialAck = true;
+        if (!m.success) {
+          initialStateError = m.error.value_or(label + " failed");
+        }
+      }
+    }
+    if (handledInitialAck) {
+      if (initialStateError.has_value() && callbacks_.onInitialStateError) {
+        callbacks_.onInitialStateError(*initialStateError);
+      }
+      return;
+    }
+  }
+
   // Acks and room_info are consumed by their waiters; everything else is an event.
   {
     std::lock_guard<std::mutex> lk(mutex_);
@@ -181,6 +245,8 @@ void SignalingChannel::onMessage(const std::string& text) {
     }
   }
 
+  if (m.type == IncomingType::PromptAck || m.type == IncomingType::SetImageAck) return;
+
   switch (m.type) {
     case IncomingType::QueuePosition: {
       {
@@ -238,16 +304,54 @@ void SignalingChannel::onSocketError(const std::string& reason) {
   cv_.notify_all();
 }
 
+void SignalingChannel::startInitialAckTimer() {
+  joinInitialAckTimer();
+  initialAckTimer_ = std::thread([this] {
+    std::optional<std::string> initialStateError;
+    {
+      std::unique_lock<std::mutex> lk(mutex_);
+      while (initialAck_.has_value() && !fatalError_ && !closing_) {
+        if (cv_.wait_until(lk, initialAck_->deadline) == std::cv_status::timeout &&
+            std::chrono::steady_clock::now() >= initialAck_->deadline) {
+          initialStateError = initialAck_->label + " timed out";
+          initialAck_.reset();
+          cv_.notify_all();
+          break;
+        }
+      }
+    }
+    if (initialStateError.has_value() && callbacks_.onInitialStateError) {
+      callbacks_.onInitialStateError(*initialStateError);
+    }
+  });
+}
+
+void SignalingChannel::joinInitialAckTimer() {
+  if (!initialAckTimer_.joinable()) return;
+  if (initialAckTimer_.get_id() == std::this_thread::get_id()) {
+    initialAckTimer_.detach();
+    return;
+  }
+  initialAckTimer_.join();
+}
+
 bool SignalingChannel::sendText(const std::string& text) {
   if (!ws_) return false;
   return ws_->sendText(text).success;
 }
 
 void SignalingChannel::close() {
+  bool alreadyClosed = false;
   {
     std::lock_guard<std::mutex> lk(mutex_);
-    if (closing_ && !ws_) return;
+    alreadyClosed = closing_ && !ws_;
     closing_ = true;
+    initialAck_.reset();
+    cv_.notify_all();
+  }
+  if (alreadyClosed) {
+    joinInitialAckTimer();
+    return;
   }
   // stop() joins the WS thread; must not hold the lock (the Close callback locks).
   if (ws_) ws_->stop();
@@ -257,6 +361,7 @@ void SignalingChannel::close() {
     if (!fatalError_) fatalError_ = "Signaling channel closed";
     cv_.notify_all();
   }
+  joinInitialAckTimer();
 }
 
 } // namespace decart::detail
diff --git a/src/detail/signaling_channel.h b/src/detail/signaling_channel.h
@@ -9,6 +9,7 @@
 #include <nlohmann/json.hpp>
 #include <optional>
 #include <string>
+#include <thread>
 #include <vector>
 
 #include "realtime/messages.h"
@@ -27,6 +28,15 @@ struct RoomInfo {
   std::string sessionId;
 };
 
+/// Initial prompt/image bundled into `livekit_join.initial_state`. The ack is
+/// observed asynchronously so media publishing does not wait on it.
+struct InitialStateAckRequest {
+  nlohmann::json message;
+  std::function<bool(const IncomingMessage&)> matchAck;
+  std::chrono::milliseconds timeout;
+  std::string label;
+};
+
 /// The Decart realtime signaling channel: a WebSocket to `wss://api3.decart.ai`
 /// that performs the join handshake, carries prompt/image control messages
 /// (request/ack), and forwards server events. Media flows separately over
@@ -43,6 +53,8 @@ class SignalingChannel {
     std::function<void(double seconds)> onGenerationTick;
     std::function<void(double seconds, const std::string& reason)> onGenerationEnded;
     std::function<void(const std::string& message)> onServerError;
+    /// Failed ack for the initial state bundled into `livekit_join`.
+    std::function<void(const std::string& message)> onInitialStateError;
     /// Unexpected close after a successful connect (code, reason).
     std::function<void(int code, const std::string& reason)> onClosed;
   };
@@ -61,7 +73,8 @@ class SignalingChannel {
   /// `connectTimeout` is a single bound covering both phases (socket open + room
   /// join). `queue_position` messages refresh the deadline, so time spent waiting
   /// in the inference queue does not count against it. @throws on failure.
-  RoomInfo openAndJoin(std::chrono::milliseconds connectTimeout);
+  RoomInfo openAndJoin(std::chrono::milliseconds connectTimeout,
+                       std::optional<InitialStateAckRequest> initialState = std::nullopt);
 
   /// Send a prompt and block until acknowledged. @throws on failure/timeout.
   void sendPrompt(const std::string& text, bool enhance, std::chrono::milliseconds timeout);
@@ -86,9 +99,17 @@ class SignalingChannel {
     bool ready = false;
   };
 
+  struct InitialAckWatch {
+    std::function<bool(const IncomingMessage&)> match;
+    std::chrono::steady_clock::time_point deadline;
+    std::string label;
+  };
+
   void onMessage(const std::string& text);
   void onClosed(int code, const std::string& reason);
   void onSocketError(const std::string& reason);
+  void startInitialAckTimer();
+  void joinInitialAckTimer();
 
   // Register `waiter`, send `message`, and block until matched or `timeout`.
   IncomingMessage request(const nlohmann::json& message, std::function<bool(const IncomingMessage&)> match,
@@ -114,6 +135,8 @@ class SignalingChannel {
   bool connected_ = false; // room_info received (handshake complete)
   bool closing_ = false;
   std::optional<std::string> fatalError_;
+  std::optional<InitialAckWatch> initialAck_;
+  std::thread initialAckTimer_;
 
   // Handshake deadline, extendable when queue_position arrives.
   std::chrono::steady_clock::time_point handshakeDeadline_{};

diff --git a/src/realtime/session.cpp b/src/realtime/session.cpp
@@ -82,6 +82,42 @@ std::string resolveImageBase64(const std::string& image) {
   return image;
 }
 
+InitialStateAckRequest buildInitialStateRequest(const InitialState& initial) {
+  const std::optional<std::string> initialPrompt =
+      initial.prompt.has_value() ? std::optional<std::string>(initial.prompt->text) : std::nullopt;
+  const std::optional<bool> initialEnhance =
+      initial.prompt.has_value() ? std::optional<bool>(initial.prompt->enhance) : std::nullopt;
+
+  if (initial.image.has_value()) {
+    if (initial.image->ref.has_value()) {
+      return InitialStateAckRequest{
+          makeSetImageRefMsg(*initial.image->ref, initialPrompt, initialEnhance),
+          [](const IncomingMessage& m) { return m.type == IncomingType::SetImageAck; }, kUpdateTimeout,
+          "Image send"};
+    }
+    if (initial.image->image.has_value()) {
+      return InitialStateAckRequest{
+          makeSetImageDataMsg(resolveImageBase64(*initial.image->image), initialPrompt, initialEnhance),
+          [](const IncomingMessage& m) { return m.type == IncomingType::SetImageAck; }, kUpdateTimeout,
+          "Image send"};
+    }
+  }
+
+  if (initial.prompt.has_value()) {
+    const std::string text = initial.prompt->text;
+    return InitialStateAckRequest{
+        makePromptMsg(text, initial.prompt->enhance),
+        [text](const IncomingMessage& m) { return m.type == IncomingType::PromptAck && m.prompt == text; },
+        kPromptTimeout, "Prompt send"};
+  }
+
+  // Match the JS SDK bootstrap: publish a null image with the join so the
+  // backend can start passthrough/generation without a post-publish gate.
+  return InitialStateAckRequest{makeSetImageDataMsg(std::nullopt, std::nullopt, std::nullopt),
+                                [](const IncomingMessage& m) { return m.type == IncomingType::SetImageAck; },
+                                kUpdateTimeout, "Image send"};
+}
+
 } // namespace
 
 /// Shared media half of a session: owns the LiveKit room, maps its events to the
@@ -472,6 +508,9 @@ std::unique_ptr<RealtimeSession> connectRealtime(std::shared_ptr<ClientConfig> c
   callbacks.onServerError = [cb = options.onError](const std::string& message) {
     if (cb) cb(Error{ErrorCode::ServerError, message});
   };
+  callbacks.onInitialStateError = [cb = options.onError](const std::string& message) {
+    if (cb) cb(Error{ErrorCode::ServerError, message});
+  };
   // An unexpected signaling drop after connect means control is dead: reflect it
   // in the connection state so isConnected() stops returning true, then notify.
   callbacks.onClosed = [impl, cb = options.onError](int code, const std::string& reason) {
@@ -483,37 +522,14 @@ std::unique_ptr<RealtimeSession> connectRealtime(std::shared_ptr<ClientConfig> c
   impl->signaling->setCallbacks(std::move(callbacks));
 
   try {
-    const RoomInfo info = impl->signaling->openAndJoin(options.connectTimeout);
-
-    // Initial reference image / prompt rides ahead of the media connect so the
-    // model has it before the first frame (mirrors the python SDK ordering).
-    bool hasInitialState = false;
-    const auto& initial = options.initialState;
-    const std::optional<std::string> initialPrompt =
-        initial.prompt.has_value() ? std::optional<std::string>(initial.prompt->text) : std::nullopt;
-    const std::optional<bool> initialEnhance =
-        initial.prompt.has_value() ? std::optional<bool>(initial.prompt->enhance) : std::nullopt;
-
-    if (initial.image.has_value()) {
-      hasInitialState = true;
-      if (initial.image->ref.has_value()) {
-        impl->signaling->setImageRef(*initial.image->ref, initialPrompt, initialEnhance, kUpdateTimeout);
-      } else if (initial.image->image.has_value()) {
-        impl->signaling->setImageData(resolveImageBase64(*initial.image->image), initialPrompt,
-                                      initialEnhance, kUpdateTimeout);
-      }
-    } else if (initial.prompt.has_value()) {
-      hasInitialState = true;
-      impl->signaling->sendPrompt(initial.prompt->text, initial.prompt->enhance, kPromptTimeout);
-    }
+    // Initial reference image / prompt rides inside livekit_join. Its ack is
+    // observed asynchronously by SignalingChannel, so track publishing is no
+    // longer gated on the prompt/image ack.
+    InitialStateAckRequest initialState = buildInitialStateRequest(options.initialState);
+    const RoomInfo info = impl->signaling->openAndJoin(options.connectTimeout, std::move(initialState));
 
     impl->media.connectRoom(info, source, options.model.fps, options.startMuted);
 
-    // With no explicit initial state, a passthrough set_image kicks off generation.
-    if (!hasInitialState) {
-      impl->signaling->setImageData(std::nullopt, std::nullopt, std::nullopt, kUpdateTimeout);
-    }
-
     impl->setSessionId(info.sessionId);
     impl->media.setState(ConnectionState::Connected);
     return session;

diff --git a/tests/test_messages.cpp b/tests/test_messages.cpp
@@ -40,6 +40,10 @@ TEST_CASE("makeJoinMsg embeds initial state or null") {
   auto withState = makeJoinMsg(makePromptMsg("hi", false));
   CHECK(withState["initial_state"]["type"] == "prompt");
   CHECK(withState["initial_state"]["prompt"] == "hi");
+
+  auto withBootstrap = makeJoinMsg(makeSetImageDataMsg(std::nullopt, std::nullopt, std::nullopt));
+  CHECK(withBootstrap["initial_state"]["type"] == "set_image");
+  CHECK(withBootstrap["initial_state"]["image_data"].is_null());
 }
 
 TEST_CASE("parseIncoming handles livekit_room_info") {