fix(pb): simplify runner wf

Author: MasterPtato

engine/packages/pegboard-runner/src/conn.rs

@@ -76,7 +76,7 @@ pub async fn init_conn(
 			}
 		};
 
-		let packet = versioned::ToServer::deserialize(&buf, protocol_version)
+		let init_packet = versioned::ToServer::deserialize(&buf, protocol_version)
 			.map_err(|err| WsError::InvalidPacket(err.to_string()).build())
 			.context("failed to deserialize initial packet from client")?;
 
@@ -86,7 +86,7 @@ pub async fn init_conn(
 				version,
 				total_slots,
 				..
-			}) = &packet
+			}) = &init_packet
 			{
 				// Look up existing runner by key
 				let existing_runner = ctx
@@ -161,12 +161,12 @@ pub async fn init_conn(
 
 				(name.clone(), runner_id, workflow_id)
 			} else {
-				tracing::debug!(?packet, "invalid initial packet");
+				tracing::debug!(?init_packet, "invalid initial packet");
 				return Err(WsError::InvalidInitialPacket("must be `ToServer::Init`").build());
 			};
 
 		// Forward to runner wf
-		ctx.signal(pegboard::workflows::runner2::Forward { inner: packet })
+		ctx.signal(pegboard::workflows::runner2::Forward { inner: init_packet })
 			.to_workflow_id(workflow_id)
 			.send()
 			.await

engine/packages/pegboard-runner/src/ws_to_tunnel_task.rs

@@ -337,9 +337,25 @@ async fn handle_message(
 				.await
 				.context("failed to handle tunnel message")?;
 		}
+		// Forward to actor wf
+		protocol::ToServer::ToServerEvents(events) => {
+			let res = ctx.signal(pegboard::workflows::runner2::Forward {
+				inner: protocol::ToServer::try_from(msg)
+					.context("failed to convert message for workflow forwarding")?,
+			})
+			.tag("actor_id", actor_id)
+			.graceful_not_found()
+			.send()
+			.await
+			.with_context(|| {
+				format!("failed to forward signal to actor workflow: {}", actor_id)
+			})?;
+			if res.is_none() {
+				tracing::warn!(?actor_id, "failed to send signal to actor workflow, likely already stopped");
+			}
+		}
 		// Forward to runner wf
 		protocol::ToServer::ToServerInit(_)
-		| protocol::ToServer::ToServerEvents(_)
 		| protocol::ToServer::ToServerAckCommands(_)
 		| protocol::ToServer::ToServerStopping => {
 			ctx.signal(pegboard::workflows::runner2::Forward {