wip

Author: Rutik Thakre

Makefile.toml

@@ -329,6 +329,20 @@ golem-cli app clean
 golem-cli app build -b ollama-debug
 '''
 
+[tasks.build-test-ollama]
+dependencies = ["build-ollama"]
+description = "Builds all test components with golem-cli"
+install_crate = "cargo-binstall"
+script = '''
+cargo-binstall golem-cli@1.2.2-dev.11 --locked --force --no-confirm
+cargo-binstall wac-cli --locked --force --no-confirm
+cd test
+golem-cli --version
+golem-cli app clean
+golem-cli app build -b ollama-debug
+'''
+
+
 [tasks.set-version]
 description = "Sets the version in all Cargo.toml files to the value of the VERSION environment variable"
 condition = { env_set = ["VERSION"] }

llm-ollama/src/lib.rs

@@ -20,7 +20,7 @@ mod conversions;
 struct OllamaChatStream {
     stream: RefCell<Option<EventSource>>,
     failure: Option<Error>,
-    finished: RefCell<bool>, 
+    finished: RefCell<bool>,
 }
 
 impl OllamaChatStream {
@@ -68,25 +68,25 @@ impl LlmChatStreamState for OllamaChatStream {
         self.stream.borrow_mut()
     }
 
-   fn decode_message(&self, raw: &str) -> Result<Option<StreamEvent>, String> {
+    fn decode_message(&self, raw: &str) -> Result<Option<StreamEvent>, String> {
         println!("OllamaChatStream::decode_message");
-        trace!("Parsing NDJSON line: {raw}");        
+        trace!("Parsing NDJSON line: {raw}");
         let json: serde_json::Value = serde_json::from_str(raw.trim())
             .map_err(|e| format!("JSON parse error: {e}"))?;
 
         if json.get("done").and_then(|v| v.as_bool()).unwrap_or(false) {
             return Ok(Some(StreamEvent::Finish(ResponseMetadata {
                 finish_reason: Some(FinishReason::Stop),
                 usage: None,
-                provider_id: None, 
+                provider_id: None,
                 timestamp: None,
                 provider_metadata_json: None,
             })));
         }
 
         if let Some(message) = json.get("message") {
             let mut content = Vec::new();
-            
+
             if let Some(text) = message.get("content").and_then(|c| c.as_str()) {
                 if !text.is_empty() {
                     content.push(ContentPart::Text(text.to_string()));

llm/src/chat_stream.rs

@@ -19,10 +19,12 @@ pub struct LlmChatStream<T> {
 
 impl<T: LlmChatStreamState> LlmChatStream<T> {
     pub fn new(implementation: T) -> Self {
+        println!("LlmChatStream::new");
         Self { implementation }
     }
 
     pub fn subscribe(&self) -> Pollable {
+        println!("LlmChatStream::subscribe");
         if let Some(stream) = self.implementation.stream().as_ref() {
             stream.subscribe()
         } else {
@@ -33,6 +35,7 @@ impl<T: LlmChatStreamState> LlmChatStream<T> {
 
 impl<T: LlmChatStreamState> GuestChatStream for LlmChatStream<T> {
     fn get_next(&self) -> Option<Vec<StreamEvent>> {
+        println!("LlmChatStream::get_next");
         if self.implementation.is_finished() {
             return Some(vec![]);
         }
@@ -99,6 +102,7 @@ impl<T: LlmChatStreamState> GuestChatStream for LlmChatStream<T> {
     }
 
     fn blocking_get_next(&self) -> Vec<StreamEvent> {
+        println!("LlmChatStream::blocking_get_next");
         let pollable = self.subscribe();
         let mut result = Vec::new();
         loop {

llm/src/event_source/event_stream.rs

@@ -167,7 +167,7 @@ impl EventStream {
     pub fn poll_next(
         &mut self,
     ) -> Poll<Option<Result<MessageEvent, EventStreamError<StreamError>>>> {
-        println!("EventStream::poll_next");
+        println!("EventStream::poll_next buffer {} ", self.buffer.as_str());
         trace!("Polling for next event");
 
         match parse_event(&mut self.buffer, &mut self.builder) {
@@ -270,18 +270,17 @@ fn parse_event<E>(
     buffer: &mut String,
     builder: &mut EventBuilder,
 ) -> Result<Option<MessageEvent>, EventStreamError<E>> {
-    println!("poll_next->parse_event");
     if buffer.is_empty() {
         return Ok(None);
     }
     loop {
-        match line(buffer.as_ref()) {
-            Ok((rem, next_line)) => {
-                builder.add(next_line);
-                let consumed = buffer.len() - rem.len();
-                let rem = buffer.split_off(consumed);
-                *buffer = rem;
-                if builder.is_complete {
+    match line(buffer.as_ref()) {
+        Ok((rem, next_line)) => {
+            builder.add(next_line);
+            let consumed = buffer.len() - rem.len();
+            let rem = buffer.split_off(consumed);
+            *buffer = rem;
+            if builder.is_complete {
                     if let Some(event) = builder.dispatch() {
                         return Ok(Some(event));
                     }