diff --git a/engines/dradis-echo/db/migrate/20260714000001_create_dradis_plugins_echo_sessions.rb b/engines/dradis-echo/db/migrate/20260714000001_create_sessions.rb
similarity index 70%
rename from engines/dradis-echo/db/migrate/20260714000001_create_dradis_plugins_echo_sessions.rb
rename to engines/dradis-echo/db/migrate/20260714000001_create_sessions.rb
index 54850df83..448504391 100644
--- a/engines/dradis-echo/db/migrate/20260714000001_create_dradis_plugins_echo_sessions.rb
+++ b/engines/dradis-echo/db/migrate/20260714000001_create_sessions.rb
@@ -1,10 +1,9 @@
-class CreateDradisPluginsEchoSessions < ActiveRecord::Migration[8.0]
+class CreateSessions < ActiveRecord::Migration[8.0]
def change
create_table :dradis_plugins_echo_sessions do |t|
t.references :agent, null: false,
foreign_key: { to_table: :dradis_plugins_echo_agents }
- t.references :user, null: true,
- foreign_key: { to_table: :users, on_delete: :nullify }
+ t.references :user, foreign_key: { to_table: :users, on_delete: :nullify }
t.references :record, polymorphic: true, null: false
# 0 maps to the :idle enum value
diff --git a/engines/dradis-echo/db/migrate/20260714000002_create_dradis_plugins_echo_messages.rb b/engines/dradis-echo/db/migrate/20260714000002_create_messages.rb
similarity index 70%
rename from engines/dradis-echo/db/migrate/20260714000002_create_dradis_plugins_echo_messages.rb
rename to engines/dradis-echo/db/migrate/20260714000002_create_messages.rb
index 3c31ed180..57e335732 100644
--- a/engines/dradis-echo/db/migrate/20260714000002_create_dradis_plugins_echo_messages.rb
+++ b/engines/dradis-echo/db/migrate/20260714000002_create_messages.rb
@@ -1,13 +1,12 @@
-class CreateDradisPluginsEchoMessages < ActiveRecord::Migration[8.0]
+class CreateMessages < ActiveRecord::Migration[8.0]
def change
create_table :dradis_plugins_echo_messages do |t|
t.references :session, null: false,
foreign_key: { to_table: :dradis_plugins_echo_sessions }
- # Self-referential parent, dormant until Slice 3 branches conversations.
+ # Self-referential parent, dormant.
t.references :parent, null: true,
foreign_key: { to_table: :dradis_plugins_echo_messages }
- t.references :user, null: true,
- foreign_key: { to_table: :users, on_delete: :nullify }
+ t.references :user, foreign_key: { to_table: :users, on_delete: :nullify }
# 0 maps to the :user role
t.integer :role, default: 0, null: false
diff --git a/engines/dradis-echo/spec/factories/messages.rb b/engines/dradis-echo/spec/factories/messages.rb
index 5a06146d0..9dd84c641 100644
--- a/engines/dradis-echo/spec/factories/messages.rb
+++ b/engines/dradis-echo/spec/factories/messages.rb
@@ -1,6 +1,7 @@
FactoryBot.define do
factory :echo_message, class: 'Dradis::Plugins::Echo::Message' do
association :session, factory: :echo_session
+ user
role { :user }
content { 'Hello, Echo.' }
From cf88f494e505c5860a684c8558e30be5a6817302 Mon Sep 17 00:00:00 2001
From: dradis-bot <286253174+dradis-bot@users.noreply.github.com>
Date: Tue, 14 Jul 2026 18:21:05 +0100
Subject: [PATCH 09/57] Echo: replace InteractionJob with ReplyJob and
generation state machine
ReplyJob streams an assistant reply into a persisted message, strips thinking blocks, records model/provider metadata, then serialises: re-enqueue if the user spoke again mid-generation, else flip the session idle and broadcast the composer state. Session#request_reply! is the idle->generating lock gate with stuck-generation recovery.
---
.../dradis/plugins/echo/interaction_job.rb | 32 -----
.../app/jobs/dradis/plugins/echo/reply_job.rb | 113 ++++++++++++++++++
.../app/models/dradis/plugins/echo/session.rb | 57 +++++++++
.../sessions/_composer_state.html.erb | 11 ++
.../sessions/messages/_message.html.erb | 3 +-
5 files changed, 183 insertions(+), 33 deletions(-)
delete mode 100644 engines/dradis-echo/app/jobs/dradis/plugins/echo/interaction_job.rb
create mode 100644 engines/dradis-echo/app/jobs/dradis/plugins/echo/reply_job.rb
create mode 100644 engines/dradis-echo/app/views/dradis/plugins/echo/projects/sessions/_composer_state.html.erb
diff --git a/engines/dradis-echo/app/jobs/dradis/plugins/echo/interaction_job.rb b/engines/dradis-echo/app/jobs/dradis/plugins/echo/interaction_job.rb
deleted file mode 100644
index 0a358f50e..000000000
--- a/engines/dradis-echo/app/jobs/dradis/plugins/echo/interaction_job.rb
+++ /dev/null
@@ -1,32 +0,0 @@
-module Dradis::Plugins::Echo
- class InteractionJob < ApplicationJob
- queue_as :dradis_project
-
- def perform(agent_id:, prompt:, interaction_id:, response_id:)
- agent = Agent.find(agent_id)
- raise "Agent '#{agent.name}' is not enabled" unless agent.enabled?
-
- spinner_shown = true
-
- agent.provider.generate(prompt: prompt, model: agent.model_override) do |chunk|
- if spinner_shown
- Turbo::StreamsChannel.broadcast_remove_to [interaction_id, 'prompts'], target: "#{response_id}_spinner"
- spinner_shown = false
- end
-
- Turbo::StreamsChannel.broadcast_append_to(
- [interaction_id, 'prompts'],
- target: response_id,
- content: ERB::Util.html_escape(chunk)
- )
- end
-
- Turbo::StreamsChannel.broadcast_append_to [interaction_id, 'prompts'], target: 'messages', html: '
Done.
'
- rescue => e
- msg = '
'
- msg << ERB::Util.html_escape(e.message)
- msg << '
'
- Turbo::StreamsChannel.broadcast_update_to [interaction_id, 'prompts'], target: response_id, html: msg
- end
- end
-end
diff --git a/engines/dradis-echo/app/jobs/dradis/plugins/echo/reply_job.rb b/engines/dradis-echo/app/jobs/dradis/plugins/echo/reply_job.rb
new file mode 100644
index 000000000..e219e22e6
--- /dev/null
+++ b/engines/dradis-echo/app/jobs/dradis/plugins/echo/reply_job.rb
@@ -0,0 +1,113 @@
+module Dradis::Plugins::Echo
+ class ReplyJob < ApplicationJob
+ queue_as :dradis_project
+
+ # Generates one assistant reply for a session: it streams the provider
+ # response into a `streaming` message, persists the final text as
+ # `complete` with model/provider metadata, then serializes — re-enqueueing
+ # itself if the user spoke again mid-generation, or flipping the session
+ # back to `idle`. Session#request_reply! owns the idle->generating gate.
+ def perform(session)
+ agent = session.agent
+ raise "Agent '#{agent.name}' is not enabled" unless agent.enabled?
+
+ cutoff_id = session.messages.maximum(:id).to_i
+ context = session.to_provider_messages
+ message = session.messages.create!(role: :assistant, status: :streaming)
+
+ text, duration_ms = stream_reply(agent, session, message, context)
+
+ complete(agent, message, text, duration_ms)
+ serialize(session, cutoff_id)
+ rescue => e
+ fail_message(session, message, e)
+ end
+
+ private
+
+ def stream_reply(agent, session, message, context)
+ buffer = +''
+ started = clock
+
+ agent.provider.generate(messages: context, model: agent.model_override) do |chunk|
+ buffer << chunk
+ broadcast_chunk(session, message, chunk)
+ end
+
+ [buffer, ((clock - started) * 1000).round]
+ end
+
+ def broadcast_chunk(session, message, chunk)
+ Turbo::StreamsChannel.broadcast_append_to(
+ [session, :messages],
+ target: ActionView::RecordIdentifier.dom_id(message, :content),
+ content: ERB::Util.html_escape(chunk)
+ )
+ end
+
+ def complete(agent, message, text, duration_ms)
+ message.update!(
+ content: strip_thinking(text),
+ status: :complete,
+ metadata: message.metadata.merge(
+ 'duration_ms' => duration_ms,
+ 'model' => agent.model_override.presence || agent.provider.model,
+ 'provider' => agent.provider.type_name
+ )
+ )
+ broadcast_message(message)
+ end
+
+ # Providers surface their reasoning either as raw
tags or,
+ # for Ollama, as the {thinking}{/thinking} markers Provider::Ollama swaps
+ # them for. Neither belongs in the persisted answer, so drop the blocks
+ # and any stray markers before saving.
+ def strip_thinking(text)
+ text
+ .gsub(/\{thinking\}.*?\{\/thinking\}/m, '')
+ .gsub(/
.*?<\/think>/m, '')
+ .gsub(/\{\/?thinking\}/, '')
+ .gsub(/<\/?think>/, '')
+ .strip
+ end
+
+ # Under a lock so it can't race the controller flipping idle<->generating:
+ # if a user message landed after the reply started (id past the cutoff),
+ # answer it too by re-enqueueing; otherwise release the session to idle.
+ def serialize(session, cutoff_id)
+ session.with_lock do
+ if session.messages.where(role: :user).where('id > ?', cutoff_id).exists?
+ self.class.perform_later(session)
+ else
+ session.update!(status: :idle)
+ session.broadcast_composer_state
+ end
+ end
+ end
+
+ def fail_message(session, message, error)
+ if message
+ message.update!(
+ status: :failed,
+ metadata: message.metadata.merge('error' => error.message)
+ )
+ broadcast_message(message)
+ end
+
+ session.with_lock { session.update!(status: :idle) }
+ session.broadcast_composer_state
+ end
+
+ def broadcast_message(message)
+ message.broadcast_replace_to(
+ [message.session, :messages],
+ partial: 'dradis/plugins/echo/projects/sessions/messages/message',
+ locals: { message: message }
+ )
+ end
+
+ def clock
+ Process.clock_gettime(Process::CLOCK_MONOTONIC)
+ end
+ end
+end
diff --git a/engines/dradis-echo/app/models/dradis/plugins/echo/session.rb b/engines/dradis-echo/app/models/dradis/plugins/echo/session.rb
index 5d1b13251..74f4f9f0f 100644
--- a/engines/dradis-echo/app/models/dradis/plugins/echo/session.rb
+++ b/engines/dradis-echo/app/models/dradis/plugins/echo/session.rb
@@ -1,5 +1,9 @@
module Dradis::Plugins::Echo
class Session < ApplicationRecord
+ # Beyond the provider read timeout a still-streaming message can only be the
+ # debris of a crashed job, so we allow this much slack before reclaiming it.
+ STUCK_MARGIN = 30.seconds
+
enum :status, %i[idle generating], default: :idle
# -- Relationships --------------------------------------------------------
@@ -27,6 +31,42 @@ def self.record_type_for(record)
end
# -- Instance Methods -----------------------------------------------------
+ # Broadcasts the composer partial so the browser reflects the current
+ # idle/generating state. Called on every idle<->generating transition.
+ def broadcast_composer_state
+ broadcast_replace_to(
+ [self, :composer_state],
+ target: ActionView::RecordIdentifier.dom_id(self, :composer_state),
+ partial: 'dradis/plugins/echo/projects/sessions/composer_state',
+ locals: { session: self }
+ )
+ end
+
+ def project
+ record.project
+ end
+
+ # The gate in front of ReplyJob: flips idle->generating and enqueues exactly
+ # one job. A no-op while already generating, so repeated calls (a user
+ # sending several messages) never double-enqueue — the running job re-checks
+ # for newer messages when it finishes. A generation whose streaming message
+ # has gone stale is treated as dead and reclaimed first.
+ def request_reply!
+ enqueue = false
+
+ with_lock do
+ reclaim_stuck_generation! if generating?
+
+ if idle?
+ update!(status: :generating)
+ broadcast_composer_state
+ enqueue = true
+ end
+ end
+
+ ReplyJob.perform_later(self) if enqueue
+ end
+
def to_provider_messages
messages.order(:created_at, :id).map do |message|
{ role: message.role, content: message.content }
@@ -44,5 +84,22 @@ def record=(new_record)
self.record_type = self.class.record_type_for(new_record) if new_record
new_record
end
+
+ private
+
+ # A crashed ReplyJob leaves the session locked in `generating` with an
+ # orphaned streaming message. Once that message hasn't been touched past the
+ # provider read timeout (plus a margin), fail it and release the session so
+ # the next request_reply! can start fresh.
+ def reclaim_stuck_generation!
+ threshold = Provider::HttpStreaming::READ_TIMEOUT.seconds.ago - STUCK_MARGIN
+ stuck = messages.where(role: :assistant, status: :streaming).where(updated_at: ..threshold)
+ return unless stuck.exists?
+
+ stuck.find_each do |message|
+ message.update!(status: :failed, metadata: message.metadata.merge('error' => 'interrupted'))
+ end
+ update!(status: :idle)
+ end
end
end
diff --git a/engines/dradis-echo/app/views/dradis/plugins/echo/projects/sessions/_composer_state.html.erb b/engines/dradis-echo/app/views/dradis/plugins/echo/projects/sessions/_composer_state.html.erb
new file mode 100644
index 000000000..68db682c4
--- /dev/null
+++ b/engines/dradis-echo/app/views/dradis/plugins/echo/projects/sessions/_composer_state.html.erb
@@ -0,0 +1,11 @@
+<%# locals: (session:) -%>
+<%# Broadcast on every idle<->generating transition (Session#broadcast_composer_state). %>
+<%# Route-free and locals-only so it renders from ReplyJob. Slice 4 wires the %>
+<%# composer form; here we only reflect whether a reply is in flight. %>
+
+ <% if session.generating? %>
+ Generating…
+ <% else %>
+ Ready
+ <% end %>
+
diff --git a/engines/dradis-echo/app/views/dradis/plugins/echo/projects/sessions/messages/_message.html.erb b/engines/dradis-echo/app/views/dradis/plugins/echo/projects/sessions/messages/_message.html.erb
index 9554bedcf..5cbf37b2f 100644
--- a/engines/dradis-echo/app/views/dradis/plugins/echo/projects/sessions/messages/_message.html.erb
+++ b/engines/dradis-echo/app/views/dradis/plugins/echo/projects/sessions/messages/_message.html.erb
@@ -12,7 +12,8 @@
<% end %>
- <%= message.content %>
+ <%# ReplyJob streams chunks into this container by dom_id(message, :content). %>
+ <%= message.content %>
<% if message.streaming? %>
From e659993a27a4d50f67f0e7f732b777950528465b Mon Sep 17 00:00:00 2001
From: dradis-bot <286253174+dradis-bot@users.noreply.github.com>
Date: Tue, 14 Jul 2026 18:21:05 +0100
Subject: [PATCH 10/57] Echo: spec ReplyJob and Session#request_reply! locking
---
.../plugins/echo/interaction_job_spec.rb | 70 ---------
.../dradis/plugins/echo/reply_job_spec.rb | 140 ++++++++++++++++++
.../dradis/plugins/echo/session_spec.rb | 51 +++++++
3 files changed, 191 insertions(+), 70 deletions(-)
delete mode 100644 engines/dradis-echo/spec/jobs/dradis/plugins/echo/interaction_job_spec.rb
create mode 100644 engines/dradis-echo/spec/jobs/dradis/plugins/echo/reply_job_spec.rb
diff --git a/engines/dradis-echo/spec/jobs/dradis/plugins/echo/interaction_job_spec.rb b/engines/dradis-echo/spec/jobs/dradis/plugins/echo/interaction_job_spec.rb
deleted file mode 100644
index 9827e9a81..000000000
--- a/engines/dradis-echo/spec/jobs/dradis/plugins/echo/interaction_job_spec.rb
+++ /dev/null
@@ -1,70 +0,0 @@
-require 'rails_helper'
-require File.expand_path('../../../../factories/agents', __dir__)
-require File.expand_path('../../../../factories/providers', __dir__)
-
-describe Dradis::Plugins::Echo::InteractionJob do
- let(:interaction_id) { 'project-1' }
- let(:response_id) { 'response-1' }
- let(:prompt) { 'Summarise this issue.' }
- let(:agent) { create(:system_agent) }
-
- def perform
- described_class.perform_now(
- agent_id: agent.id,
- prompt: prompt,
- interaction_id: interaction_id,
- response_id: response_id
- )
- end
-
- before do
- allow(Turbo::StreamsChannel).to receive(:broadcast_append_to)
- allow(Turbo::StreamsChannel).to receive(:broadcast_remove_to)
- allow(Turbo::StreamsChannel).to receive(:broadcast_update_to)
- end
-
- describe 'when agent is not enabled' do
- before { agent.update!(enabled: false) }
-
- it 'broadcasts a user-friendly error' do
- perform
- expect(Turbo::StreamsChannel).to have_received(:broadcast_update_to) do |_, **kwargs|
- expect(kwargs[:html]).to include('is not enabled')
- end
- end
- end
-
- describe 'error message sanitisation' do
- it 'HTML-escapes the error message before broadcasting' do
- allow_any_instance_of(Dradis::Plugins::Echo::Provider::Ollama)
- .to receive(:generate).and_raise('')
-
- perform
- expect(Turbo::StreamsChannel).to have_received(:broadcast_update_to) do |_, **kwargs|
- expect(kwargs[:html]).to include('<script>')
- expect(kwargs[:html]).not_to include('')
+
+ perform
+ expect(Turbo::StreamsChannel).to have_received(:broadcast_update_to) do |_, **kwargs|
+ expect(kwargs[:html]).to include('<script>')
+ expect(kwargs[:html]).not_to include('')
-
- perform
- expect(Turbo::StreamsChannel).to have_received(:broadcast_update_to) do |_, **kwargs|
- expect(kwargs[:html]).to include('<script>')
- expect(kwargs[:html]).not_to include('