middleware_messaging.md

Middleware Messaging Guide

This guide explains the middleware messaging pattern in Sagents, which enables targeted communication with middleware running inside an AgentServer — both from external processes (LiveViews, controllers) and from async tasks spawned by the middleware itself.

Table of Contents

• Overview
• External Notifications
• Architecture
• Core Concepts
• Implementation Guide
• API Reference
• Examples
• Best Practices
• Troubleshooting

Overview

The middleware messaging pattern allows:

1. External processes (LiveViews, controllers) to send targeted updates to a specific middleware in a running AgentServer
2. Middleware to spawn async tasks and receive results back via the same mechanism
3. State updates in response to messages, which are then available to before_model/2 on the next LLM call
4. Broadcast updates to subscribers (LiveViews, external clients)

All messages are routed through AgentServer.notify_middleware/3, which delivers them to the target middleware's handle_message/3 callback.

Two primary use cases

External notifications — A LiveView or controller sends context to a middleware that the middleware couldn't know on its own. For example, telling a "user context" middleware which document the user is currently viewing, so the middleware can inject that context into the next LLM message.

Async task results — A middleware spawns a background task (e.g., title generation, embedding computation) and the task sends results back to the middleware for state updates.

Both use cases share the same API and routing mechanism — the distinction is just who calls notify_middleware/3.

External Notifications

This is the most common pattern for application code interacting with middleware at runtime. It allows your LiveViews, controllers, or other processes to send targeted messages to a specific middleware in a running AgentServer.

Why not try to update metadata directly?

A generic "update metadata" API is untargeted — it dumps data into a shared bucket without the middleware knowing about it. notify_middleware/3 routes directly to a specific middleware by ID, letting the middleware:

• Validate the message
• Decide how to store it (metadata, config, side effects)
• Maintain its own invariants
• Pattern match on specific message types

Pattern: LiveView → Middleware context updates

# In your LiveView — user navigated to a different document
def handle_event("select_document", %{"id" => id}, socket) do
  doc = Writing.get_document!(socket.assigns.current_scope, id)

  if socket.assigns.agent_id do
    AgentServer.notify_middleware(
      socket.assigns.agent_id,
      MyApp.Middleware.UserContext,
      {:current_document_changed, %{path: doc.path, title: doc.title}}
    )
  end

  {:noreply, assign(socket, :active_document, doc)}
end

# In the middleware — handle the notification
@impl true
def handle_message({:current_document_changed, doc_info}, state, _config) do
  {:ok, State.put_metadata(state, "current_document", doc_info)}
end

# In before_model — use the stored context
@impl true
def before_model(state, _config) do
  case State.get_metadata(state, "current_document") do
    nil -> {:ok, state}
    %{path: path, title: title} ->
      context = "Currently viewing: #{title} (#{path})"
      {:ok, prepend_to_last_user_message(state, context)}
  end
end

Pattern: Preference or setting changes

# Toggle verbose mode from a settings panel
AgentServer.notify_middleware(agent_id, MyApp.Middleware.Preferences, {:set, :verbose, true})

# In the middleware
@impl true
def handle_message({:set, key, value}, state, _config) do
  prefs = State.get_metadata(state, "preferences") || %{}
  {:ok, State.put_metadata(state, "preferences", Map.put(prefs, key, value))}
end

Key characteristics

• Fire-and-forget: notify_middleware/3 returns :ok immediately. The caller doesn't wait for a response.
• Safe when agent is down: If the AgentServer isn't running, the message is silently dropped.
• Targeted: The message is routed to a specific middleware by ID — not broadcast to all middleware.
• Middleware owns interpretation: The middleware's handle_message/3 decides what to do with the message. It can update metadata, trigger side effects, or ignore it.

Architecture

High-Level Message Flow

There are two flows — external notifications and async task results — that converge on the same handle_message/3 callback:

sequenceDiagram
    participant LV as LiveView/Controller
    participant AS as AgentServer
    participant M as Middleware
    participant T as Async Task
    participant PS as PubSub

    Note over LV,M: Flow 1: External notification
    LV->>AS: notify_middleware(agent_id, middleware_id, message)
    AS->>M: handle_message(message, state, config)
    M-->>AS: {:ok, updated_state}

    Note over LV,M: Flow 2: Async task result
    AS->>M: after_model(state, config)
    M->>T: Task.start(fn -> ...)
    M-->>AS: {:ok, state}
    T->>T: Complete work
    T->>AS: notify_middleware(agent_id, middleware_id, result)
    AS->>M: handle_message(result, state, config)
    M-->>AS: {:ok, updated_state}
    AS->>PS: Broadcast debug event (state update)

Loading

Middleware Lifecycle with Messaging

stateDiagram-v2
    [*] --> Init: Agent new
    Init --> BeforeModel: User message arrives
    BeforeModel --> LLMCall: Middleware transforms state
    LLMCall --> AfterModel: LLM response ready
    AfterModel --> CheckCondition: after_model callback

    CheckCondition --> SpawnTask: Condition met
    CheckCondition --> Return: Condition not met

    SpawnTask --> CaptureServerPID: Get self()
    CaptureServerPID --> StartTask: Task start
    StartTask --> Return: Task runs in background
    Return --> [*]: Execution continues

    state AsyncTask {
        [*] --> DoWork: Task executing
        DoWork --> Success: Work completes
        DoWork --> Failure: Error occurs
        Success --> SendSuccess: send message
        Failure --> SendFailure: send error
        SendSuccess --> [*]
        SendFailure --> [*]
    }

    state MessageHandling {
        [*] --> ReceiveMessage: handle_info called
        ReceiveMessage --> RouteMessage: Lookup middleware by ID
        RouteMessage --> HandleMessage: handle_message callback
        HandleMessage --> UpdateState: Modify state
        UpdateState --> Broadcast: Broadcast requested
        Broadcast --> [*]: PubSub notified
    }

Loading

Core Concepts

1. Middleware Identification

Each middleware instance has a unique identifier used for message routing:

• Default ID: The module name (e.g., Sagents.Middleware.ConversationTitle)
• Custom ID: Specified via :id option to support multiple instances

# Single instance - uses module name as ID
middleware = [
  {ConversationTitle, [
    chat_model: model
  ]}
]

# Multiple instances - custom IDs required
middleware = [
  {ConversationTitle, [
    id: "english_title",
    chat_model: model,
    prompt_template: "Generate English title..."
  ]},
  {ConversationTitle, [
    id: "spanish_title",
    chat_model: model,
    prompt_template: "Generate Spanish title..."
  ]}
]

2. Sending Messages with notify_middleware/3

All middleware messages — whether from external processes or async tasks — use the same public API:

AgentServer.notify_middleware(agent_id, middleware_id, message)

Where:

• agent_id: The agent identifier string
• middleware_id: Atom (module name) or string (custom ID)
• message: Any term (typically a tagged tuple like {:result, data})

From external processes (LiveViews, controllers):

# User navigated to a new document — tell the middleware
AgentServer.notify_middleware(agent_id, MyApp.UserContext, {:document_changed, doc_info})

From async tasks spawned by middleware:

def after_model(state, config) do
  # Capture agent_id before spawning task
  agent_id = state.agent_id
  middleware_id = Map.get(config, :id, __MODULE__)

  Task.start(fn ->
    result = do_work()
    AgentServer.notify_middleware(agent_id, middleware_id, {:result, result})
  end)

  {:ok, state}
end

3. State Updates and Broadcasting

The handle_message/3 callback updates state and can trigger broadcasts manually:

def handle_message({:result, data}, state, _config) do
  updated_state = State.put_metadata(state, "key", data)

  # The state update is automatically applied by AgentServer
  # To broadcast events to subscribers, use AgentServer.publish_event_from/2
  # (typically done in the async task before sending the message)
  {:ok, updated_state}
end

Middleware can broadcast custom events using AgentServer.publish_event_from/2:

# In async task after work completes
AgentServer.publish_event_from(agent_id, {:custom_event, data})
AgentServer.notify_middleware(agent_id, middleware_id, {:result, data})

When state is updated via handle_message/3, a debug event is automatically broadcast:

• {:agent_state_update, middleware_id, updated_state} - For debugging/monitoring

Implementation Guide

Step 1: Define the Middleware Module

defmodule MyApp.Middleware.CustomMiddleware do
  @behaviour Sagents.Middleware

  require Logger
  alias Sagents.State

  @impl true
  def init(opts) do
    # Extract and validate configuration
    api_key = Keyword.get(opts, :api_key)

    unless api_key do
      {:error, "CustomMiddleware requires :api_key option"}
    else
      {:ok, %{api_key: api_key}}
    end
  end
end

Step 2: Implement Lifecycle Hooks

@impl true
def after_model(state, config) do
  # Check if async work is needed
  if should_process?(state) do
    spawn_async_task(state, config)
  end

  {:ok, state}
end

defp should_process?(state) do
  # Custom logic to determine if task should run
  State.get_metadata(state, "processed") == nil
end

Step 3: Spawn Async Tasks

defp spawn_async_task(state, config) do
  # IMPORTANT: Capture variables from parent context before spawning
  agent_id = state.agent_id

  # Get middleware ID for routing
  middleware_id = Map.get(config, :id, __MODULE__)

  # Extract data needed for processing
  data = extract_data(state)

  # Emit telemetry
  :telemetry.execute(
    [:sagents, :middleware, :task, :spawned],
    %{count: 1},
    %{middleware: middleware_id, task_type: :custom_processing}
  )

  Task.start(fn ->
    try do
      # Do the actual work
      result = process_data(data, config)

      # Emit success telemetry
      :telemetry.execute(
        [:sagents, :middleware, :task, :completed],
        %{count: 1},
        %{middleware: middleware_id, task_type: :custom_processing}
      )

      # Broadcast events to subscribers (if needed)
      AgentServer.publish_event_from(agent_id, {:custom_event, result})

      # Send success message back to AgentServer
      AgentServer.notify_middleware(agent_id, middleware_id, {:success, result})
    rescue
      error ->
        Logger.error("Processing failed: #{inspect(error)}")

        # Emit failure telemetry
        :telemetry.execute(
          [:sagents, :middleware, :task, :failed],
          %{count: 1},
          %{middleware: middleware_id, task_type: :custom_processing, error: inspect(error)}
        )

        # Send failure message back
        AgentServer.notify_middleware(agent_id, middleware_id, {:error, error})
    end
  end)
end

Step 4: Handle Messages

@impl true
def handle_message({:success, result}, state, _config) do
  Logger.info("Processing completed: #{inspect(result)}")

  # Update state with result
  updated_state =
    state
    |> State.put_metadata("processed", true)
    |> State.put_metadata("result", result)

  # State is automatically updated by AgentServer
  # Events should be broadcast from the async task using AgentServer.publish_event_from/2
  {:ok, updated_state}
end

def handle_message({:error, error}, state, _config) do
  Logger.error("Processing failed: #{inspect(error)}")

  # Don't update state, just log the error
  {:ok, state}
end

Step 5: Use the Middleware

{:ok, agent} = Agent.new(
  agent_id: "my-agent",
  model: chat_model,
  middleware: [
    {MyApp.Middleware.CustomMiddleware, [
      api_key: "secret-key"
    ]}
  ]
)

API Reference

Middleware Behavior Callbacks

init/1

@callback init(options :: keyword()) ::
  {:ok, config :: map()}
  | {:error, reason :: term()}

Initialize the middleware with configuration options. Called once during agent creation.

Parameters:

• options - Keyword list of configuration options

Returns:

• {:ok, config} - Configuration map that will be passed to other callbacks
• {:error, reason} - Initialization failed

handle_message/3

@callback handle_message(message :: term(), State.t(), config :: map()) ::
  {:ok, State.t()}
  | {:error, reason :: term()}

Handle messages sent from async tasks.

Parameters:

• message - The message payload (typically a tagged tuple)
• state - Current agent state
• config - Middleware configuration from init/1

Returns:

• {:ok, state} - Updated state
• {:error, reason} - Message handling failed (error is logged but does not halt agent execution)

Note: To broadcast events to subscribers, use AgentServer.publish_event_from/2 from within the async task before sending the message, rather than attempting to broadcast from handle_message/3.

on_server_start/2

@callback on_server_start(State.t(), config :: map()) ::
  {:ok, State.t()}
  | {:error, reason :: term()}

Called when the AgentServer starts or restarts.

This allows middleware to perform initialization actions that require the AgentServer to be running, such as broadcasting initial state to subscribers (e.g., TODOs for UI display).

Parameters:

• state - The current agent state
• config - Middleware configuration from init/1

Returns:

• {:ok, state} - Success (state typically unchanged but can be modified)
• {:error, reason} - Error (logged but does not halt agent startup)

Example:

@impl true
def on_server_start(state, _config) do
  # Broadcast initial todos when AgentServer starts
  AgentServer.publish_event_from(state.agent_id, {:todos_updated, state.todos})
  {:ok, state}
end

AgentServer.notify_middleware/3

Send a targeted message to a specific middleware in a running AgentServer. Works from any process — LiveViews, controllers, async tasks, or other GenServers.

AgentServer.notify_middleware(agent_id, middleware_id, message)

Parameters:

• agent_id - The agent identifier string
• middleware_id - Module name (atom) or custom ID (string) of the target middleware
• message - Any term to be delivered to the middleware's handle_message/3 callback

Returns: :ok (always — fire-and-forget, silent no-op if agent is not running)

Note: notify_middleware/3 is strongly preferred over raw send/2 because it:

• Provides a clearer, more maintainable interface
• Handles the case where the AgentServer may not be running
• Is consistent with other AgentServer operations
• Makes testing easier by providing a clear seam for mocking

Telemetry Events

The following telemetry events are emitted:

[:sagents, :middleware, :task, :spawned]

Emitted when an async task is spawned.

Measurements: %{count: 1} Metadata: %{middleware: id, task_type: atom}

[:sagents, :middleware, :task, :completed]

Emitted when a task completes successfully.

Measurements: %{count: 1} Metadata: %{middleware: id, task_type: atom}

[:sagents, :middleware, :task, :failed]

Emitted when a task fails.

Measurements: %{count: 1} Metadata: %{middleware: id, task_type: atom, error: string}

[:sagents, :middleware, :message, :received]

Emitted when AgentServer receives a middleware message.

Measurements: %{count: 1} Metadata: %{middleware_id: id, agent_id: string}

Examples

Example 1: ConversationTitle Middleware

The built-in ConversationTitle middleware demonstrates the complete pattern:

defmodule Sagents.Middleware.ConversationTitle do
  @behaviour Sagents.Middleware

  require Logger
  alias Sagents.{State, AgentServer}
  alias LangChain.Chains.TextToTitleChain

  @impl true
  def init(opts) do
    chat_model = Keyword.get(opts, :chat_model)

    unless chat_model do
      {:error, "ConversationTitle middleware requires :chat_model option"}
    else
      {:ok, %{
        chat_model: chat_model,
        fallbacks: Keyword.get(opts, :fallbacks, []),
        examples: Keyword.get(opts, :examples, default_examples())
      }}
    end
  end

  @impl true
  def before_model(state, config) do
    # Only generate title if we don't have one yet
    if State.get_metadata(state, "conversation_title") == nil do
      spawn_title_generation_task(state, config)
    end

    {:ok, state}
  end

  @impl true
  def handle_message({:title_generated, title}, state, _config) do
    Logger.debug("Received title: #{title}")
    updated_state = State.put_metadata(state, "conversation_title", title)
    {:ok, updated_state}
  end

  def handle_message({:title_generation_failed, reason}, state, _config) do
    Logger.warning("Title generation failed: #{inspect(reason)}")
    {:ok, state}
  end

  defp spawn_title_generation_task(state, config) do
    # Capture agent_id for API calls
    agent_id = state.agent_id
    middleware_id = Map.get(config, :id, __MODULE__)
    user_text = extract_last_user_message_text(state)

    # Emit telemetry
    :telemetry.execute(
      [:sagents, :middleware, :task, :spawned],
      %{count: 1},
      %{middleware: middleware_id, task_type: :title_generation}
    )

    Task.start(fn ->
      try do
        # Generate title
        title = generate_title(user_text, config)

        # Emit telemetry
        :telemetry.execute(
          [:sagents, :middleware, :task, :completed],
          %{count: 1},
          %{middleware: middleware_id, task_type: :title_generation}
        )

        # Broadcast event to subscribers
        AgentServer.publish_event_from(agent_id, {:conversation_title_generated, title, agent_id})

        # Send message to AgentServer for state update
        AgentServer.notify_middleware(agent_id, middleware_id, {:title_generated, title})
      rescue
        error ->
          Logger.error("Title generation failed: #{inspect(error)}")

          :telemetry.execute(
            [:sagents, :middleware, :task, :failed],
            %{count: 1},
            %{middleware: middleware_id, task_type: :title_generation, error: inspect(error)}
          )

          AgentServer.notify_middleware(agent_id, middleware_id, {:title_generation_failed, error})
      end
    end)
  end

  defp generate_title(text, config) do
    TextToTitleChain.new!(%{
      llm: config.chat_model,
      input_text: text,
      examples: config.examples
    })
    |> TextToTitleChain.evaluate(with_fallbacks: config.fallbacks)
  end

  defp extract_last_user_message_text(state) do
    # Extract text from the last user message
    state.messages
    |> Enum.reverse()
    |> Enum.find(fn msg -> msg.role == :user end)
    |> case do
      nil -> ""
      message -> LangChain.Message.ContentPart.parts_to_string(message.content)
    end
  end
end

Usage:

{:ok, agent} = Sagents.new(
  model: main_model,
  middleware: [
    {Sagents.Middleware.ConversationTitle, [
      chat_model: ChatAnthropic.new!(%{model: "claude-3-5-haiku-latest"}),
      fallbacks: [backup_model]
    ]}
  ]
)

Example 2: Analytics Middleware

Track conversation analytics asynchronously:

defmodule MyApp.Middleware.Analytics do
  @behaviour Sagents.Middleware

  require Logger
  alias Sagents.{State, AgentServer}

  @impl true
  def init(opts) do
    {:ok, %{
      analytics_api: Keyword.get(opts, :analytics_api),
      track_tokens: Keyword.get(opts, :track_tokens, true)
    }}
  end

  @impl true
  def after_model(state, config) do
    # Send analytics in background
    spawn_analytics_task(state, config)
    {:ok, state}
  end

  @impl true
  def handle_message({:analytics_sent, event_id}, state, _config) do
    Logger.debug("Analytics event sent: #{event_id}")
    {:ok, state}
  end

  def handle_message({:analytics_failed, reason}, state, _config) do
    Logger.warning("Analytics failed: #{inspect(reason)}")
    {:ok, state}
  end

  defp spawn_analytics_task(state, config) do
    agent_id = state.agent_id
    middleware_id = Map.get(config, :id, __MODULE__)

    # Extract analytics data
    message_count = length(state.messages)

    Task.start(fn ->
      try do
        event_id = send_analytics(agent_id, message_count, config)
        AgentServer.notify_middleware(agent_id, middleware_id, {:analytics_sent, event_id})
      rescue
        error ->
          AgentServer.notify_middleware(agent_id, middleware_id, {:analytics_failed, error})
      end
    end)
  end

  defp send_analytics(agent_id, message_count, config) do
    # Make API call to analytics service
    config.analytics_api.track_event(%{
      agent_id: agent_id,
      message_count: message_count,
      timestamp: DateTime.utc_now()
    })
  end
end

Example 3: Document Embedding Middleware

Generate embeddings for conversation history:

defmodule MyApp.Middleware.Embeddings do
  @behaviour Sagents.Middleware

  require Logger
  alias Sagents.{State, AgentServer}

  @impl true
  def init(opts) do
    {:ok, %{
      embedding_model: Keyword.get(opts, :embedding_model),
      batch_size: Keyword.get(opts, :batch_size, 10)
    }}
  end

  @impl true
  def after_model(state, config) do
    # Check if we should generate embeddings
    message_count = length(state.messages)

    if rem(message_count, config.batch_size) == 0 do
      spawn_embedding_task(state, config)
    end

    {:ok, state}
  end

  @impl true
  def handle_message({:embeddings_generated, embeddings}, state, _config) do
    Logger.info("Generated #{length(embeddings)} embeddings")

    updated_state = State.put_metadata(state, "embeddings", embeddings)
    {:ok, updated_state}
  end

  def handle_message({:embeddings_failed, reason}, state, _config) do
    Logger.error("Embedding generation failed: #{inspect(reason)}")
    {:ok, state}
  end

  defp spawn_embedding_task(state, config) do
    agent_id = state.agent_id
    middleware_id = Map.get(config, :id, __MODULE__)

    # Extract text from messages
    texts = extract_message_texts(state)

    Task.start(fn ->
      try do
        embeddings = generate_embeddings(texts, config)
        AgentServer.notify_middleware(agent_id, middleware_id, {:embeddings_generated, embeddings})
      rescue
        error ->
          AgentServer.notify_middleware(agent_id, middleware_id, {:embeddings_failed, error})
      end
    end)
  end

  defp generate_embeddings(texts, config) do
    # Call embedding API
    config.embedding_model.embed(texts)
  end

  defp extract_message_texts(state) do
    Enum.map(state.messages, fn msg ->
      LangChain.Message.ContentPart.parts_to_string(msg.content)
    end)
  end
end

Best Practices

1. Always Capture agent_id Early

# ✅ GOOD - Capture agent_id from state
def after_model(state, config) do
  agent_id = state.agent_id  # Captured before spawning task
  middleware_id = Map.get(config, :id, __MODULE__)

  Task.start(fn ->
    # Use agent_id with public API
    AgentServer.notify_middleware(agent_id, middleware_id, {:result, data})
  end)

  {:ok, state}
end

# ❌ BAD - Trying to access state inside task
def after_model(state, config) do
  Task.start(fn ->
    agent_id = state.agent_id  # state is not in task closure!
    AgentServer.notify_middleware(agent_id, ...)
  end)

  {:ok, state}
end

2. Use Descriptive Message Tags

# ✅ GOOD - Clear message intent
AgentServer.notify_middleware(agent_id, id, {:title_generated, title})
AgentServer.notify_middleware(agent_id, id, {:title_generation_failed, error})

# ❌ BAD - Ambiguous messages
AgentServer.notify_middleware(agent_id, id, title)
AgentServer.notify_middleware(agent_id, id, {:error, error})

3. Include Error Handling

# ✅ GOOD - Comprehensive error handling
Task.start(fn ->
  try do
    result = do_work()
    AgentServer.notify_middleware(agent_id, id, {:success, result})
  rescue
    error ->
      Logger.error("Task failed: #{inspect(error)}")
      AgentServer.notify_middleware(agent_id, id, {:error, error})
  end
end)

# ❌ BAD - No error handling
Task.start(fn ->
  result = do_work()  # Crash if this fails!
  AgentServer.notify_middleware(agent_id, id, {:success, result})
end)

4. Emit Telemetry Events

# ✅ GOOD - Track task lifecycle
:telemetry.execute([:sagents, :middleware, :task, :spawned], %{count: 1}, metadata)
# ... do work ...
:telemetry.execute([:sagents, :middleware, :task, :completed], %{count: 1}, metadata)

# This enables monitoring, alerting, and debugging

5. Broadcast Events Strategically

# ✅ GOOD - Broadcast from async task when event is significant
Task.start(fn ->
  result = do_work()

  # Broadcast to subscribers if this is a significant event
  if significant_change?(result) do
    AgentServer.publish_event_from(agent_id, {:important_update, result})
  end

  # Always send message to update state
  AgentServer.notify_middleware(agent_id, middleware_id, {:result, result})
end)

# State updates via handle_message/3 automatically trigger debug events
# Custom events for user-facing updates should be published explicitly

6. Handle Task Timeouts

Consider using Task.Supervisor for production deployments:

defp spawn_with_timeout(agent_id, middleware_id, work_fn) do
  Task.Supervisor.start_child(MyApp.TaskSupervisor, fn ->
    try do
      # Work with timeout
      result = work_fn.()
      AgentServer.notify_middleware(agent_id, middleware_id, {:success, result})
    catch
      :exit, {:timeout, _} ->
        AgentServer.notify_middleware(agent_id, middleware_id, {:timeout, "Task exceeded timeout"})
    end
  end)
end

7. Design for Multiple Instances

If your middleware might be used multiple times, support custom IDs:

@impl true
def init(opts) do
  # Accept custom ID
  custom_id = Keyword.get(opts, :id)

  config = %{
    # ... other config ...
  }

  # Store custom ID if provided
  config = if custom_id, do: Map.put(config, :id, custom_id), else: config

  {:ok, config}
end

defp spawn_task(state, config) do
  agent_id = state.agent_id
  # Use custom ID if available, otherwise module name
  middleware_id = Map.get(config, :id, __MODULE__)

  Task.start(fn ->
    result = do_work()
    AgentServer.notify_middleware(agent_id, middleware_id, {:result, result})
  end)
end

Troubleshooting

Messages Not Being Received

Problem: Async task completes but handle_message/3 is never called.

Solutions:

1. Verify you captured agent_id = state.agent_id before spawning the task
2. Check the middleware ID matches between spawn and message send
3. Ensure you're using AgentServer.notify_middleware(agent_id, middleware_id, message)
4. Verify the AgentServer is still running (check with AgentServer.get_status/1)
5. Check logs for any crashes in the AgentServer or task

Events Not Broadcasting

Problem: State updates but subscribers don't receive notifications.

Solutions:

1. Ensure you're calling AgentServer.publish_event_from/2 from the async task (not from handle_message/3)
2. Verify the AgentServer was started with PubSub configured
3. Check subscribers are properly subscribed using AgentServer.subscribe/1
4. Verify agent_id is correct when calling publish_event_from/2
5. Look for broadcast errors in logs

Tasks Hanging or Not Completing

Problem: Tasks seem to hang indefinitely.

Solutions:

1. Add timeout to async operations
2. Use Task.Supervisor for better task management
3. Add logging to track task progress
4. Check for infinite loops or deadlocks in task code
5. Monitor telemetry events to detect stuck tasks

Multiple Middleware Instances Conflicting

Problem: Multiple instances of the same middleware interfere with each other.

Solutions:

1. Use custom IDs via :id option
2. Ensure each instance has a unique ID
3. Check message routing logic handles custom IDs
4. Verify state updates don't conflict (use different metadata keys)

Memory Leaks from Tasks

Problem: Memory usage grows over time.

Solutions:

1. Ensure all spawned tasks eventually complete or timeout
2. Use Task.Supervisor with proper restart strategies
3. Clean up resources in task error handlers
4. Monitor task count with telemetry
5. Avoid capturing large state structures in task closures

See Also

• ConversationTitle Middleware Implementation
• TodoList Middleware Implementation
• Middleware Behavior Definition
• MiddlewareEntry Struct
• AgentServer Implementation