diff --git a/src/detail/signaling_channel.cpp b/src/detail/signaling_channel.cpp index b035901..87c03f0 100644 --- 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 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 initialMessage = + initialState.has_value() ? std::optional(std::move(initialState->message)) + : std::nullopt; + if (!sendText(makeJoinMsg(initialMessage).dump())) { std::lock_guard 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 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 serialize(requestMutex_); + std::optional initialStateError; + { + std::unique_lock 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->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 initialStateError; + bool handledInitialAck = false; + { + std::lock_guard 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 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 initialStateError; + { + std::unique_lock 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 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 index f7ce8de..5af0ccd 100644 --- a/src/detail/signaling_channel.h +++ b/src/detail/signaling_channel.h @@ -9,6 +9,7 @@ #include #include #include +#include #include #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 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 onGenerationTick; std::function onGenerationEnded; std::function onServerError; + /// Failed ack for the initial state bundled into `livekit_join`. + std::function onInitialStateError; /// Unexpected close after a successful connect (code, reason). std::function 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 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 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 match, @@ -114,6 +135,8 @@ class SignalingChannel { bool connected_ = false; // room_info received (handshake complete) bool closing_ = false; std::optional fatalError_; + std::optional 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 index a30fb14..017e709 100644 --- 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 initialPrompt = + initial.prompt.has_value() ? std::optional(initial.prompt->text) : std::nullopt; + const std::optional initialEnhance = + initial.prompt.has_value() ? std::optional(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 connectRealtime(std::shared_ptr 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 connectRealtime(std::shared_ptr 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 initialPrompt = - initial.prompt.has_value() ? std::optional(initial.prompt->text) : std::nullopt; - const std::optional initialEnhance = - initial.prompt.has_value() ? std::optional(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 index 1aae3c7..d053801 100644 --- 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") {