WIP Stash

backend/requirements.txt

@@ -5,7 +5,7 @@ sqlalchemy
 nest_asyncio
 replicate
 qdrant-client[fastembed]
-ollama
+#ollama
 tortoise-orm>=1.0.0
-llama-cpp-python
+#llama-cpp-python
 openrouter

backend/src/agent.py

@@ -2,6 +2,7 @@ import re
 import inspect
 import time
 import asyncio
+import traceback
 
 from temporary import generate_tmp_id
 from stream import Stream
@@ -66,7 +67,18 @@ class Agent:
             },
         )
         await self.run(run)
-        return run.get_output_stream()
+        out_stream = Stream(self.app)
+
+        def on_output_event(stream):
+            def on_chunk(chunk):
+                out_stream.feed(chunk)
+            stream.listen(on_chunk, lambda: None)
+        def on_done(data):
+            out_stream.close()
+        run.on('output', on_output_event)
+        run.on('done', on_done)
+
+        return out_stream
 
     async def received_message(self, message):
         chat_view = await self.app.get_chat_view(message.chat_id)
@@ -213,7 +225,8 @@ class AgentRun:
                 cb(data)
             except Exception as e:
                 print(f"ERROR: handling event '{event_name}' in run {self.id}\n" + str(e))
-        
+                print(traceback.format_exc())
+
         # Broadcast event to websocket listeners
         self.app.emit_event("run-event", {
             "runId": self.id,
@@ -221,24 +234,6 @@ class AgentRun:
             "data": data
         })
     
-    def get_output_stream(self):
-        if not self.output:
-            self.output = Stream(self.app)
-            self.emit_event('output', self.output)
-        return self.output
-    
-    def feed_output(self, chunk):
-        stream = self.get_output_stream()
-        if chunk:
-            self.output.feed(chunk)
-    
-    def feed_reasoning(self, chunk):
-        if not self.reasoning:
-            self.reasoning = Stream(self.app)
-            self.emit_event('reasoning', self.reasoning)
-        if chunk:
-            self.reasoning.feed(chunk)
-    
     def add_tools(self, tools):
         if "tools" in self.input:
             self.input["tools"] += tools
@@ -298,7 +293,7 @@ class AgentRun:
         self.close()
         
         # Emit cancellation event
-        self.emit_event("cancelled", {"runId": self.id, "agentId": self.agent_id})
+        self.emit_event("cancelled", {})
         
         # Broadcast cancellation event
         self.app.emit_event("run-cancelled", {
@@ -314,4 +309,4 @@ class AgentRun:
         self.end_time = int(time.time())
         if self.output: self.output.close()
         if self.reasoning: self.reasoning.close()
-        self.emit_event("done", {"output": self.output.content, "agent_id": self.agent_id})
+        self.emit_event("done", {})

backend/src/agents/expert.py

@@ -1,6 +1,8 @@
 import inference
 from agent import Agent, AgentField
 
+import asyncio
+
 CLASS_DESCRIPTION = """
 Expert AI with customizable system prompt.
 Will get guidelines, documents, and other advanced features in the future.
@@ -22,16 +24,18 @@ class ExpertAgent(Agent):
     async def run(self, run):
         opts = {}
 
-        agent_tools = self.tools
-        print('agent tools ' + repr(agent_tools))
-        run.add_tools(agent_tools)
+        using_tools = False
+        if 'tools' not in run.input:
+            if self.tools:
+                run.input['tools'] = self.tools
+                using_tools = True
 
         if self.config.get("temperature") is not None:
             opts["temperature"] = self.config["temperature"]
         if self.config.get("repetition_penalty") is not None:
             opts["repetition_penalty"] = self.config["repetition_penalty"]
 
-        # 1. Build message list based on input
+        # Build message list based on input
         messages = inference.build_messages(
             system_prompt=self.config.get("system_prompt", ""),
             instruction=run.input.get("instruction"),
@@ -39,14 +43,44 @@ class ExpertAgent(Agent):
             messages=run.input.get("messages"),
         )
 
-        # 2. Call inference, passing the run to feed run.stream and tool calls
-        await inference.chat_completion(
-            self.app,
-            model_id=self.config.get("model"),
-            messages=messages,
-            tools=run.input.get("tools", []),
-            opts=opts,
-            agent_run=run,
-        )
+        # Collect inference output. These are resused across calls
+        tool_calls = []
+        output_stream = None
+        inference_end = asyncio.Event()
+        def on_output(stream):
+            nonlocal output_stream
+            output_stream = stream
+        run.on('output', on_output)
+        run.on('inference-end', lambda event: inference_end.set())
+        run.on('tool-call', lambda tc: tool_calls.append(tc))
 
-        run.get_output_stream().listen(lambda chunk: None, lambda: run.close())
+        while True:
+            # Reset data
+            inference_end.clear()
+            tool_calls = []
+            output_stream = None
+
+            # Call inference, passing the run to get events out
+            await inference.chat_completion(
+                self.app,
+                model_id=self.config.get("model"),
+                messages=messages,
+                tools=run.input.get("tools", []),
+                opts=opts,
+                agent_run=run,
+            )
+
+            await inference_end.wait()
+
+            messages.append(inference.LlamaMessage('assistant', output_stream and await output_stream.content_wait() or ""))
+
+            if using_tools and tool_calls:
+                for tc in tool_calls:
+                    tool = next(t for t in self.tools if t.name == tc.tool_name)
+                    if not tool:
+                        print(f"Bad agent tool name: {tc.name}")
+                        continue
+                    tc.result = await tool.execute_async(**tc.arguments)
+                    messages.append(inference.LlamaMessage('tool', tc))
+            else:
+                break

backend/src/inference.py

@@ -2,12 +2,12 @@ import threading
 import asyncio
 import json
 import os
-import ollama
+#import ollama
 import traceback
 from stream import Stream
 from tool import format_tools_for_engine, build_tool_prompt_prefix, ToolSpec, ToolCall
 
-from llama_cpp import Llama
+#from llama_cpp import Llama
 from openrouter import OpenRouter
 
 VALID_ROLES = ['user', 'assistant', 'system', 'tool']
@@ -35,12 +35,60 @@ class Models:
     qwen3_8b = "lucataco/qwen3-vl-8b-instruct:39e893666996acf464cff75688ad49ac95ef54e9f1c688fbc677330acc478e11"
     qwen3_235b = "qwen/qwen3-235b-a22b-instruct-2507"
 
+def format_openai(messages):
+    formatted = []
+    last_assistant_message = None
+    for msg in messages:
+        if msg.role == 'tool':
+            if not last_assistant_message:
+                raise Exception("tool message before any assistant message")
+            tool_calls = last_assistant_message.get('tool_calls')
+            if tool_calls == None:
+                tool_calls = []
+                last_assistant_message['tool_calls'] = tool_calls
+            tool_calls.append({
+                'id': msg.content.id,
+                'type': 'function',
+                'function': {
+                    'name': msg.content.tool_name,
+                    'arguments': json.dumps(msg.content.arguments),
+                },
+            })
+            formatted.append({
+                'role': 'tool',
+                'tool_call_id': msg.content.id,
+                'content': json.dumps(msg.content.output),
+            })
+        else:
+            fmsg = {'role': msg.role, 'content': msg.content}
+            formatted.append(fmsg)
+            if msg.role == 'assistant':
+                last_assistant_message = fmsg
+    return formatted
+
+
 # llama cpp models
 model_cache = {}
 last_model_file = None
 last_model_instance = None
 
+def safe_queue(callback):
+    queue = asyncio.Queue()
+    loop = asyncio.get_running_loop()
+    async def read_loop():
+        while True:
+            args = await queue.get()
+            await callback(*args)
+    loop.create_task(read_loop())
+
+    def send(*args):
+        def queue_put():
+            loop.create_task(queue.put(args))
+        loop.call_soon_threadsafe(queue_put)
+    return send
+
 def get_llama_cpp_model(app, model_file):
+    raise Exception("llama-cpp disabled")
     global last_model_file, last_model_instance
     if last_model_file == model_file:
         return last_model_instance
@@ -95,24 +143,13 @@ def build_messages(system_prompt="", instruction=None, input=None, messages=None
 
 async def chat_completion(app,
     model_id=None,
-    instruction=None,
-    input=None,
     messages=None,
-    system_prompt="",
     tools=[],
     opts={},
     agent_run=None
 ):
     db_model = (await app.get_model(model_id)).value
 
-    # Integrate arguments into message list
-    messages = build_messages(
-        system_prompt=system_prompt,
-        instruction=instruction,
-        input=input,
-        messages=messages,
-    )
-
     model_engine = db_model['engine']
     model_name = db_model['model']
 
@@ -140,10 +177,22 @@ async def chat_completion(app,
 
 # llama.cpp
 async def local_chat_completion(app, messages, model, tools=[], opts={}, agent_run=None):
-    if agent_run:
-        content_stream = agent_run.get_output_stream()
-    else:
-        content_stream = Stream(app)
+    if not agent_run:
+        raise Exception("agent_run required")
+
+    content_stream = None
+    async def message_loop(name, value=None):
+        nonlocal content_stream, agent_run
+        if name == 'output':
+            if content_stream == None:
+                content_stream = Stream(app)
+                agent_run.emit_event('output', content_stream)
+            content_stream.feed(value)
+        elif name == 'close':
+            if content_stream:
+                content_stream.close()
+            agent_run.emit_event('inference-end', {})
+    send_message = safe_queue(message_loop)
 
     def worker():
         llm = get_llama_cpp_model(app, model)
@@ -179,21 +228,16 @@ async def local_chat_completion(app, messages, model, tools=[], opts={}, agent_r
                 
             delta = chunk['choices'][0]['delta']
             if 'content' in delta:
-                chunk_text = delta['content']
-                content_stream.feed(chunk_text)
+                send_message('output', delta['content'])
             # Handle tool_call deltas if the model emits them
             if 'tool_calls' in delta:
                 # Forward tool call chunks as JSON strings for the consumer
-                for tc in delta['tool_calls']:
-                    content_stream.feed(json.dumps({'tool_call': tc}))
-                if agent_run:
-                    for tc in delta['tool_calls']:
-                        agent_run.emit_event("tool_call", {
-                            "id": tc.get("id"),
-                            "function": tc.get("function"),
-                        })
+                send_message("tool-call", {
+                    "id": tc.get("id"),
+                    "function": tc.get("function"),
+                })
 
-        content_stream.close()
+        send_message('close')
 
     thread = threading.Thread(target=worker)
     if agent_run:
@@ -209,6 +253,7 @@ async def ollama_list(app, messages, model, opts={}):
     return ollama.list()
 
 async def ollama_chat_completion(app, messages, model, tools=[], opts={}, agent_run=None):
+    raise Exception("ollama disabled")
     if agent_run:
         content_stream = agent_run.stream
     else:
@@ -259,6 +304,29 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
     if not agent_run:
         raise Exception("agent_run required")
 
+    # TODO: Separate as InferenceRun class
+    content_stream = None
+    reasoning_stream = None
+    async def message_loop(name, value=None):
+        nonlocal content_stream, reasoning_stream, agent_run
+        if name == 'output':
+            if content_stream == None:
+                content_stream = Stream(app)
+                agent_run.emit_event('output', content_stream)
+            content_stream.feed(value)
+        if name == 'reasoning':
+            if reasoning_stream == None:
+                reasoning_stream = Stream(app)
+                agent_run.emit_event('output', reasoning_stream)
+            reasoning_stream.feed(value)
+        if name == 'tool-call':
+            agent_run.emit_event('tool-call', value)
+        elif name == 'close':
+            if content_stream: content_stream.close()
+            if reasoning_stream: reasoning_stream.close()
+            agent_run.emit_event('inference-end', {})
+    emit = safe_queue(message_loop)
+
     async def worker():
         print("START OPENROUTER INFERENCE")
         if not app.openrouter_client:
@@ -266,7 +334,7 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
 
         kwargs = {
             'model': model,
-            'messages': [msg.to_dict() for msg in messages],
+            'messages': format_openai(messages),
             'stream': True,
         }
 
@@ -294,9 +362,9 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
                     delta = event.choices[0].delta
 
                     if delta.content:
-                        agent_run.feed_output(delta.content)
+                        emit('output', delta.content)
                     if delta.reasoning:
-                        agent_run.feed_reasoning(delta.reasoning)
+                        emit('reasoning', delta.reasoning)
                     if delta.tool_calls:
                         print("[openrouter] tool call:" + repr(delta.tool_calls)) 
                         for tc in delta.tool_calls:
@@ -325,9 +393,9 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
                 try:
                     arguments = json.loads(tc.arguments)
                     tc.arguments = arguments
-                    agent_run.emit_event('tool-call', tc)
+                    emit('tool-call', tc)
                 except Exception:
-                    print("[openrouter] Bad tool call arguments")
+                    print("[openrouter] Bad tool call arguments for " + repr(tc))
                     print(traceback.format_exc())
         except asyncio.CancelledError:
             print(f"[openrouter] task cancelled for run {agent_run.id}")
@@ -336,7 +404,7 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
             print(f"[openrouter] exception during inference")
             print(traceback.format_exc())
 
-        agent_run.close()
+        emit('close')
         print("END OPENROUTER INFERENCE")
 
     task = asyncio.create_task(worker())

backend/src/migrations/0001_initial.py

@@ -1,6 +1,6 @@
 from tortoise import migrations
 from tortoise.migrations import operations as ops
-from orjson import loads
+from json import loads
 from tortoise.fields.base import OnDelete
 from tortoise.fields.data import JSON_DUMPS
 from tortoise import fields

backend/src/server.py

@@ -587,4 +587,4 @@ async def start_server(app):
         }
 
 
-    server.run(debug=True)
+    await server.run_task(debug=True)

backend/src/tool.py

@@ -1,4 +1,5 @@
 import inspect
+import asyncio
 from typing import get_type_hints
 
 class ToolParameter:
@@ -225,7 +226,8 @@ class ToolCall:
         self.result = None
 
     def __repr__(self):
-        return f"ToolCall(tool={self.tool_name!r}, args={self.arguments!r})"
+        id_prefix = self.id + ', ' if self.id else ''
+        return f"ToolCall({id_prefix}tool={self.tool_name!r}, args={self.arguments!r})"
 
     def call_message(self):
         raise Exception("TODO: Llama message for tool call")