fix(gateway): wake WebChat writes on the lws service thread

src/gateway/http.c

@@ -163,6 +163,9 @@ void http_stop(void)
 		g_ctx->thread = 0;
 	}
 	if (g_ctx->lws_ctx) {
+		/* Drop the pointer before destroy so a late send cannot
+		 * lws_cancel_service a freed context. */
+		ws_set_context(NULL);
 		lws_context_destroy(g_ctx->lws_ctx);
 		g_ctx->lws_ctx = NULL;
 	}

src/gateway/http_lws.c

@@ -488,6 +488,10 @@ int ws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user,
 		if (conn_id > 0) ws_unregister_conn(conn_id);
 		break;
 	}
+	case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
+		/* wsi here is not a client socket. Mark real conns from our table. */
+		ws_on_service_cancel();
+		break;
 	default:
 		break;
 	}

src/gateway/ws.c

@@ -8,7 +8,31 @@
 #ifdef SHELLCLAW_WS_TEST
 struct lws;
 struct lws_context;
-static void lws_callback_on_writable(struct lws *wsi) { (void)wsi; }
+static int g_ws_test_writable_count;
+static int g_ws_test_cancel_count;
+static void lws_callback_on_writable(struct lws *wsi)
+{
+	(void)wsi;
+	g_ws_test_writable_count++;
+}
+static void lws_cancel_service(struct lws_context *ctx)
+{
+	(void)ctx;
+	g_ws_test_cancel_count++;
+}
+int ws_test_writable_count(void)
+{
+	return g_ws_test_writable_count;
+}
+int ws_test_cancel_count(void)
+{
+	return g_ws_test_cancel_count;
+}
+void ws_test_reset_wake_counts(void)
+{
+	g_ws_test_writable_count = 0;
+	g_ws_test_cancel_count = 0;
+}
 #else
 #include "gateway/lws_compat.h"
 #include <libwebsockets.h>
@@ -59,7 +83,48 @@ static struct lws_context *g_lws_ctx;
 
 void ws_set_context(void *ctx)
 {
+	pthread_mutex_lock(&g_mutex);
 	g_lws_ctx = (struct lws_context *)ctx;
+	pthread_mutex_unlock(&g_mutex);
+}
+
+/* Cross-thread wake. lws_cancel_service is the call that is safe off the
+ * service thread; the writable mark happens in ws_on_service_cancel. */
+static void ws_wake_service(void)
+{
+	struct lws_context *ctx;
+
+	pthread_mutex_lock(&g_mutex);
+	ctx = g_lws_ctx;
+	pthread_mutex_unlock(&g_mutex);
+	if (ctx)
+		lws_cancel_service(ctx);
+}
+
+void ws_on_service_cancel(void)
+{
+	struct lws *wake[MAX_CONNECTIONS];
+	int n = 0;
+	int i;
+
+	pthread_mutex_lock(&g_mutex);
+	for (i = 0; i < MAX_CONNECTIONS; i++) {
+		ws_msg_t *msg;
+		int conn_id;
+
+		if (!g_conns[i].wsi)
+			continue;
+		conn_id = g_conns[i].conn_id;
+		for (msg = g_outgoing_head; msg; msg = msg->next) {
+			if (msg->conn_id != conn_id)
+				continue;
+			wake[n++] = (struct lws *)g_conns[i].wsi;
+			break;
+		}
+	}
+	pthread_mutex_unlock(&g_mutex);
+	for (i = 0; i < n; i++)
+		lws_callback_on_writable(wake[i]);
 }
 
 int ws_next_conn_id(void)
@@ -250,15 +315,14 @@ int ws_send_to(const char *session_id, const char *text)
 	}
 	pthread_mutex_unlock(&g_mutex);
 	if (!wsi) return 0;
-	lws_callback_on_writable((struct lws *)wsi);
+	ws_wake_service();
 	return 0;
 }
 
 void ws_broadcast_text(const char *text)
 {
 	int i;
-	int wake_count = 0;
-	ws_conn_t wake_list[MAX_CONNECTIONS];
+	int queued = 0;
 	if (!ws_text_len_ok(text)) return;
 	pthread_mutex_lock(&g_mutex);
 	for (i = 0; i < MAX_CONNECTIONS; i++) {
@@ -284,16 +348,11 @@ void ws_broadcast_text(const char *text)
 			g_outgoing_tail = m;
 		}
 		g_outgoing_count++;
-		if (wake_count < MAX_CONNECTIONS) {
-			wake_list[wake_count] = g_conns[i].wsi;
-			wake_count++;
-		}
+		queued = 1;
 	}
 	pthread_mutex_unlock(&g_mutex);
-	for (i = 0; i < wake_count; i++) {
-		if (wake_list[i])
-			lws_callback_on_writable((struct lws *)wake_list[i]);
-	}
+	if (queued)
+		ws_wake_service();
 }
 
 void ws_shutdown_signal(void)

src/gateway/ws.h

@@ -97,6 +97,17 @@ int ws_dequeue_outgoing(int conn_id, char *buf, size_t buf_size, size_t *len_out
  */
 int ws_has_pending_outgoing(int conn_id);
 
+/**
+ * Ask the service thread to flush queued WebChat frames.
+ * Call only from the libwebsockets thread, in
+ * LWS_CALLBACK_EVENT_WAIT_CANCELLED. lws_callback_on_writable() is not
+ * safe from the agent thread: libwebsockets 4.3 updates the poll set
+ * without the per-thread lock.
+ *
+ * Example: `case LWS_CALLBACK_EVENT_WAIT_CANCELLED: ws_on_service_cancel(); break;`
+ */
+void ws_on_service_cancel(void);
+
 /**
  * Get next connection ID for new connections.
  *

tests/test_ws.c

@@ -268,6 +268,68 @@ static int test_pop_incoming_invalid_args(void)
 	return 0;
 }
 
+/* Counters live in the SHELLCLAW_WS_TEST build of ws.c. */
+extern int ws_test_writable_count(void);
+extern int ws_test_cancel_count(void);
+extern void ws_test_reset_wake_counts(void);
+
+/**
+ * Agent replies run on the main thread while lws_service runs on the gateway
+ * thread. The wake must be lws_cancel_service, and only the service-thread
+ * hook may call lws_callback_on_writable.
+ */
+static int test_send_wakes_service_thread(void)
+{
+	char buf[32];
+	size_t len;
+
+	ws_cleanup();
+	ws_test_reset_wake_counts();
+	ws_set_context((void *)(intptr_t)1);
+	ASSERT(ws_register_conn(11, (ws_conn_t)(intptr_t)11) == 0);
+	ASSERT(ws_send_to("webchat:11", "hello") == 0);
+	ASSERT(ws_test_cancel_count() == 1);
+	ASSERT(ws_test_writable_count() == 0);
+	ws_on_service_cancel();
+	ASSERT(ws_test_writable_count() == 1);
+	ASSERT(ws_dequeue_outgoing(11, buf, sizeof(buf), &len) == 1);
+	ASSERT(strcmp(buf, "hello") == 0);
+	ws_on_service_cancel();
+	ASSERT(ws_test_writable_count() == 1);
+	ws_set_context(NULL);
+	ws_cleanup();
+	return 0;
+}
+
+static int test_broadcast_wakes_once(void)
+{
+	ws_cleanup();
+	ws_test_reset_wake_counts();
+	ws_set_context((void *)(intptr_t)1);
+	ASSERT(ws_register_conn(12, (ws_conn_t)(intptr_t)12) == 0);
+	ASSERT(ws_register_conn(13, (ws_conn_t)(intptr_t)13) == 0);
+	ws_broadcast_text("status");
+	ASSERT(ws_test_cancel_count() == 1);
+	ASSERT(ws_test_writable_count() == 0);
+	ws_on_service_cancel();
+	ASSERT(ws_test_writable_count() == 2);
+	ws_set_context(NULL);
+	ws_cleanup();
+	return 0;
+}
+
+static int test_send_without_context_does_not_touch_lws(void)
+{
+	ws_cleanup();
+	ws_test_reset_wake_counts();
+	ASSERT(ws_register_conn(14, (ws_conn_t)(intptr_t)14) == 0);
+	ASSERT(ws_send_to("webchat:14", "queued") == 0);
+	ASSERT(ws_test_cancel_count() == 0);
+	ASSERT(ws_test_writable_count() == 0);
+	ws_cleanup();
+	return 0;
+}
+
 static int test_shutdown_stops_pop(void)
 {
 	char session[32];
@@ -297,6 +359,9 @@ int main(void)
 	RUN(test_push_incoming_rejects_empty());
 	RUN(test_pop_incoming_invalid_args());
 	RUN(test_shutdown_stops_pop());
+	RUN(test_send_wakes_service_thread());
+	RUN(test_broadcast_wakes_once());
+	RUN(test_send_without_context_does_not_touch_lws());
 	printf("test_ws: all tests passed\n");
 	return 0;
 }