Packages
fnord
0.8.74
0.9.40
0.9.39
0.9.38
0.9.37
0.9.36
0.9.35
0.9.34
0.9.33
0.9.32
0.9.31
0.9.30
0.9.29
0.9.28
0.9.27
0.9.26
0.9.25
0.9.24
0.9.23
0.9.22
0.9.21
0.9.20
0.9.19
0.9.18
0.9.17
0.9.16
0.9.15
0.9.14
0.9.13
0.9.12
0.9.11
0.9.10
0.9.9
0.9.8
0.9.7
0.9.6
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.99
0.8.98
0.8.97
0.8.96
0.8.95
0.8.94
0.8.93
0.8.92
0.8.91
0.8.90
0.8.89
0.8.88
0.8.87
0.8.86
0.8.85
0.8.84
0.8.83
0.8.82
0.8.81
0.8.80
0.8.79
0.8.78
0.8.77
0.8.76
0.8.75
0.8.74
0.8.73
0.8.72
0.8.71
0.8.70
0.8.69
0.8.68
0.8.67
0.8.66
0.8.65
0.8.64
0.8.63
0.8.62
0.8.61
0.8.60
0.8.59
0.8.58
0.8.57
0.8.56
0.8.55
0.8.54
0.8.53
0.8.52
0.8.51
0.8.50
0.8.49
0.8.48
0.8.47
0.8.46
0.8.45
0.8.44
0.8.43
0.8.42
0.8.41
0.8.40
0.8.39
0.8.38
0.8.37
0.8.36
0.8.35
0.8.34
0.8.33
0.8.32
0.8.31
0.8.30
0.8.29
0.8.27
0.8.26
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.1
0.8.0
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.3
0.7.2
0.7.1
0.7.0
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.1
0.6.0
0.5.9
0.5.8
0.5.7
0.5.6
0.5.5
0.5.4
0.5.3
0.5.2
0.5.1
0.5.0
0.4.44
0.4.43
0.4.42
0.4.41
0.4.40
0.4.39
0.4.38
0.4.37
0.4.36
0.4.35
0.4.34
0.4.33
0.4.32
0.4.30
0.4.29
0.4.28
0.4.27
0.4.26
0.4.25
0.4.24
0.4.23
0.4.22
0.4.21
0.4.20
0.4.19
0.4.18
0.4.17
0.4.16
0.4.15
0.4.14
0.4.13
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.0
0.2.0
0.1.0
AI code archaeology
Current section
Files
Jump to
Current section
Files
lib/services/conversation/interrupts.ex
defmodule Services.Conversation.Interrupts do
@moduledoc """
Queue for injecting user messages into a conversation mid-completion.
This GenServer stores a FIFO list of pending injected user messages per
conversation pid. AI.Completion will drain and apply these messages at safe
checkpoints before sending a model request or between tool-call rounds.
Additionally, it supports temporarily blocking interrupts for a conversation
during critical phases (e.g., finalization). When blocked, attempts to
interrupt should be rejected at the UI layer; this server tracks blocked state
so callers can check/decide behavior.
"""
use GenServer
@type msg :: AI.Util.msg()
@type state :: %{
queues: %{optional(pid()) => [msg]},
blocked: MapSet.t()
}
@doc """
Start the interrupt queue server.
"""
@spec start_link(Keyword.t()) :: GenServer.on_start()
def start_link(opts \\ []) do
GenServer.start_link(__MODULE__, %{}, Keyword.merge([name: __MODULE__], opts))
end
@doc """
Enqueue an injected user message for the given conversation pid.
"""
@spec request(pid(), String.t()) :: :ok
def request(conversation_pid, content)
when is_pid(conversation_pid) and
is_binary(content) do
msg = AI.Util.user_msg("[User Interjection] " <> content)
GenServer.cast(__MODULE__, {:enqueue, conversation_pid, msg})
:ok
end
@doc """
Drain all pending injected messages for a conversation.
Returns an empty list if none are pending.
"""
@spec take_all(pid()) :: [msg]
def take_all(conversation_pid) when is_pid(conversation_pid) do
GenServer.call(__MODULE__, {:take_all, conversation_pid})
end
@doc """
Returns true if any interrupts are pending for the conversation pid.
"""
@spec pending?(pid()) :: boolean()
def pending?(conversation_pid) when is_pid(conversation_pid) do
GenServer.call(__MODULE__, {:pending, conversation_pid})
end
# ---------------------------------------------------------------------------
# GenServer callbacks
# ---------------------------------------------------------------------------
@impl true
@spec init(map()) :: {:ok, map()}
def init(_state), do: {:ok, %{queues: %{}, blocked: MapSet.new()}}
@impl true
@spec handle_cast({:enqueue, pid(), msg}, state) :: {:noreply, state}
def handle_cast({:enqueue, pid, msg}, %{queues: queues} = state) do
new_queues = Map.update(queues, pid, [msg], fn list -> list ++ [msg] end)
{:noreply, %{state | queues: new_queues}}
end
@impl true
@spec handle_cast({:block, pid()}, state) :: {:noreply, state}
def handle_cast({:block, pid}, %{blocked: blocked} = state) do
{:noreply, %{state | blocked: MapSet.put(blocked, pid)}}
end
@impl true
@spec handle_cast({:unblock, pid()}, state) :: {:noreply, state}
def handle_cast({:unblock, pid}, %{blocked: blocked} = state) do
{:noreply, %{state | blocked: MapSet.delete(blocked, pid)}}
end
@impl true
@spec handle_call({:take_all, pid()}, GenServer.from(), state) :: {:reply, [msg], state}
def handle_call({:take_all, pid}, _from, %{queues: queues} = state) do
{msgs, new_queues} = Map.pop(queues, pid, [])
{:reply, msgs, %{state | queues: new_queues}}
end
@impl true
@spec handle_call({:pending, pid()}, GenServer.from(), state) :: {:reply, boolean(), state}
def handle_call({:pending, pid}, _from, %{queues: queues} = state) do
{:reply, Map.has_key?(queues, pid), state}
end
@impl true
@spec handle_call({:blocked?, pid()}, GenServer.from(), state) :: {:reply, boolean(), state}
def handle_call({:blocked?, pid}, _from, %{blocked: blocked} = state) do
{:reply, MapSet.member?(blocked, pid), state}
end
# ---------------------------------------------------------------------------
# Public API for blocking
# ---------------------------------------------------------------------------
@doc """
Block interrupts for a given conversation pid.
"""
@spec block(pid()) :: :ok
def block(conversation_pid) when is_pid(conversation_pid) do
GenServer.cast(__MODULE__, {:block, conversation_pid})
end
@doc """
Unblock interrupts for a given conversation pid.
"""
@spec unblock(pid()) :: :ok
def unblock(conversation_pid) when is_pid(conversation_pid) do
GenServer.cast(__MODULE__, {:unblock, conversation_pid})
end
@doc """
Return true if interrupts are currently blocked for conversation pid.
"""
@spec blocked?(pid()) :: boolean()
def blocked?(conversation_pid) when is_pid(conversation_pid) do
GenServer.call(__MODULE__, {:blocked?, conversation_pid})
end
end