Class: Assistant::PlanOrchestrator

Inherits:
Object
  • Object
show all
Defined in:
app/services/assistant/plan_orchestrator.rb

Overview

Runs multi-step execution plans after +declare_plan+ halts the main chat.
Lives alongside ChatService; mutates the same conversation and streams via
the parent service's callbacks.

Plan execution is one sequential state machine; splitting only for metrics
would scatter persistence, streaming, and cost-cap logic.
rubocop:disable Metrics/ClassLength, Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity

Constant Summary collapse

PLAN_STEP_MAX_RETRIES =

A single hung streaming request must not be able to consume an entire plan
step's wall-clock budget. The global RubyLLM config retries 5× at a 120s
request timeout — a 720s worst case that EXCEEDS MAX_PLAN_STEP_DURATION (600s),
so a dead Gemini SSE stream silently eats the whole step and the briefing loses
that step's data with zero salvage. (Prod conv 3388, 2026-06-15: step 2 made 9
SQL calls in 33s, then a dead stream retried for ~9.4 min until the 600s step
Timeout fired.) Capping retries here bounds one request's full retry cycle to
120s × (2 + 1) = 360s — comfortably under the step budget — so a doomed request
fails fast and the step degrades gracefully instead of timing out the briefing.

2
ASSEMBLY_MAX_TOKENS =

Total assembly budget. On OpenRouter this is a SHARED pool — reasoning
draws from it before content — so it has to hold bounded reasoning AND a
full briefing, leaving ~24K for prose after the cap above.

32_768

Instance Method Summary collapse

Constructor Details

#initialize(conversation:, chat_service:) ⇒ PlanOrchestrator

Returns a new instance of PlanOrchestrator.



23
24
25
26
 
# File 'app/services/assistant/plan_orchestrator.rb', line 23

def initialize(conversation:, chat_service:)
  @conversation = conversation
  @chat_service = chat_service
end

Instance Method Details

#build_plan_error_result(message) ⇒ Assistant::ChatService::Result

Used from ChatService for halt paths that never entered plan execution.

Parameters:

  • message (String) —

    user-facing fallback text

Returns:



259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
 
# File 'app/services/assistant/plan_orchestrator.rb', line 259

def build_plan_error_result(message)
  response_time = Process.clock_gettime(Process::CLOCK_MONOTONIC) - chat_service_ivar(:@start_time)
  chat_service_send(:assign_full_response_if_blank!, message)

  streamer = chat_service_ivar(:@streamer)
  streamer&.call(chat_service_ivar(:@full_response))
  @conversation.add_message(role: :assistant, content: chat_service_ivar(:@full_response))

  ChatService::Result.new(
    content: chat_service_ivar(:@full_response),
    model: chat_service_ivar(:@model_key),
    input_tokens: 0,
    output_tokens: 0,
    cached_tokens: 0,
    cache_creation_tokens: 0,
    response_time: response_time,
    model_reason: chat_service_ivar(:@model_selection_reason),
    thinking_text: nil,
    tool_limit_reached: false,
    tool_stats: {}
  )
end

#execute!(streamer_proc) ⇒ Assistant::ChatService::Result

Runs the persisted execution plan and assembles the user-facing reply.

Parameters:

  • streamer_proc (Proc) —

    streaming callback for the assembly turn

Returns:



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
 
# File 'app/services/assistant/plan_orchestrator.rb', line 32

def execute!(streamer_proc)
  @conversation.reload
  chat_service_send(:reset_full_response_for_plan!, +'')
  plan = @conversation.metadata['execution_plan']
  return build_plan_error_result('Plan data was missing after declaration. Please try again.') unless plan.is_a?(Hash) && plan['steps'].is_a?(Array) && plan['steps'].any?

  goal = plan['goal'].to_s
  steps = plan['steps']
  total_in = 0
  total_out = 0
  total_cached = 0
  plan_cost = 0.0
  step_results = []
  plan_tool_limit_reached = false
  merged_tool_stats = { total_tool_calls: 0, plan_steps: steps.size }
  silent_step_streamer = proc { |_| }
  @current_plan_step_index = nil

  plan_sticky_keys = Assistant::ChatToolBuilder.sticky_service_keys_for_plan(
    steps, chat_service_ivar(:@tool_services), goal: goal
  )
  cross_step_cache = {}
  # SQL HYGIENE: a single Set shared across every step's ToolLoopGuard so
  # a describe_available_data call in step 1 still satisfies the gate when
  # SQL referencing the same view runs in a later step. Without this, the
  # model would be forced to re-describe each step and burn budget.
  #
  # Seeded from earlier TURNS on the same terms ChatService uses — views
  # whose column list is still in context — so a plan doesn't re-describe
  # what the conversation already has in front of it.
  shared_described_views = Assistant::ToolLoopGuard.previously_described_views(chat_service_ivar(:@conversation))
  # Services whose tools were successfully invoked in earlier steps stay
  # available for later ones. Steps in a plan are usually correlated, and
  # a successful prior tool call is a far stronger signal that a service
  # is relevant than a regex match against the step description. Without
  # this, conversation 1657 lost blog_management on step 3 even though
  # steps 1 and 2 had just edited the same blog post with it.
  services_used_so_far = []

  steps.each_with_index do |step, idx|
    @current_plan_step_index = idx
    if chat_service_ivar(:@cancel_check)&.call
      remaining_steps = steps[idx..]
      remaining_steps.each_with_index do |_remaining_step, offset|
        persist_plan_step_metadata!(step_index: idx + offset, result_text: 'Cancelled by user', status: 'skipped')
      end
      break
    end

    persist_plan_step_metadata!(step_index: idx, result_text: nil, status: 'in_progress')
    chat_service_send(:emit_status, { event: :plan_step_start, step_index: idx, step_count: steps.size })

    service_keys = Assistant::ChatToolBuilder.tool_service_keys_for_step(
      step['description'],
      chat_service_ivar(:@tool_services),
      sticky_keys: (plan_sticky_keys + services_used_so_far).uniq
    )
    step_tools = Assistant::ChatToolBuilder.tools_for_services(
      service_keys,
      role: chat_service_ivar(:@user_role),
      allowed_objects: chat_service_ivar(:@allowed_objects),
      audit_context: {
        conversation_id: @conversation.id,
        user_id: @conversation.user_id,
        active_services: chat_service_ivar(:@tool_services)
      },
      account: chat_service_ivar(:@account),
      provider: chat_service_ivar(:@model_config)[:provider],
      include_plan_tools: false
    )
    if step_tools.empty?
      step_tools = Assistant::ChatToolBuilder.tools_for_services(
        chat_service_ivar(:@tool_services),
        role: chat_service_ivar(:@user_role),
        allowed_objects: chat_service_ivar(:@allowed_objects),
        audit_context: {
          conversation_id: @conversation.id,
          user_id: @conversation.user_id,
          active_services: chat_service_ivar(:@tool_services)
        },
        account: chat_service_ivar(:@account),
        provider: chat_service_ivar(:@model_config)[:provider],
        include_plan_tools: false
      )
    end

    step_guard = Assistant::ToolLoopGuard.new(
      role: chat_service_ivar(:@user_role),
      conversation_id: @conversation.id,
      plan_step_mode: true,
      cross_step_cache: cross_step_cache,
      supports_thinking: chat_service_ivar(:@model_config)&.dig(:supports_thinking) == true,
      cancel_check: chat_service_ivar(:@cancel_check),
      available_tool_names: step_tools.map(&:name),
      described_views: shared_described_views,
      limits_enabled: @conversation.tool_limits_enabled?
    )
    step_guard.apply!(step_tools)

    prior_payload = step_results.map do |sr|
      { step: sr[:step], description: sr[:description], result: sr[:result] }
    end

    text, in_t, out_t = run_plan_step_executor(
      step: step,
      step_index: idx,
      goal: goal,
      prior_results: prior_payload,
      step_tools: step_tools,
      streamer_proc: silent_step_streamer
    )
    text = clamp_step_result_for_chain(text)

    # What the step ACTUALLY executed. The narrative alone is not evidence:
    # conversation 4348 (2026-07-31) reported three Microsoft Ads
    # optimizations as applied — specific ad group ids, before/after bids,
    # even quoted `status: ok, CpcBid: 3.1` responses — and not one write
    # tool was ever called. All three steps were recorded `completed`
    # because the old rule below asked only whether the text was non-blank
    # and didn't start with "Step failed:".
    # nil (guard reported nothing) is UNKNOWN and must not downgrade a step;
    # only a guard that positively reports an empty list proves the step
    # called nothing. Getting this backwards would fail every step in every
    # plan the moment stats were unavailable.
    executed_tools = executed_tool_names_for(step_guard)

    step_status = if text.blank? && in_t.zero? && out_t.zero?
                    'skipped'
                  elsif text.start_with?('Step failed:', 'STEP FAILED') || text.blank?
                    'failed'
                  # Spelled out rather than `&.empty?`: nil is UNKNOWN and
                  # must NOT take this branch, but `blank?` would.
                  elsif executed_tools && executed_tools.empty?
                    # A step that called nothing changed nothing, whatever it
                    # says. Rewritten to the STEP FAILED shape so
                    # step_result_failed? / all_step_results_failed? and the
                    # assembly's FAILURE NOTICE all pick it up.
                    text = unexecuted_step_result(step, text)
                    'failed'
                  else
                    'completed'
                  end

    # Stamp the executed tools onto the result the AssemblyAgent reads, so
    # it can never see only the model's own account of the step. Also the
    # only durable record of what a step did: step agents run on their own
    # chats, so their tool calls are not persisted as assistant_tool_calls
    # and "what did it actually do" was previously unanswerable.
    if step_status == 'completed' && executed_tools.present?
      text = annotate_executed_tools(text, executed_tools)
      text = annotate_retrieved_records(text, step_guard.stats&.dig(:retrieved_records))
    end
    persist_plan_step_metadata!(step_index: idx, result_text: text, status: step_status)

    step_results << { step: idx, description: step['description'].to_s, result: text }
    total_in += in_t
    total_out += out_t

    step_cost = Assistant::CostCalculator.cost_for(
      chat_service_ivar(:@model_key),
      input_tokens: in_t,
      output_tokens: out_t
    )
    plan_cost += step_cost
    if step_guard.stats
      merged_tool_stats[:total_tool_calls] += step_guard.stats[:total_tool_calls]
      step_used = Array(step_guard.stats[:unique_tool_names])
                  .filter_map { |n| Assistant::ChatToolBuilder.service_for_tool(n) }
                  .uniq & chat_service_ivar(:@tool_services).to_a
      services_used_so_far = (services_used_so_far + step_used).uniq
    end
    plan_tool_limit_reached ||= step_guard.limit_reached?

    if plan_cost > ChatService::MAX_PLAN_COST_USD
      chat_service_send(:emit_status, { event: :plan_cost_cap, estimated_usd: plan_cost.round(4) })
      remaining = steps[(idx + 1)..]
      remaining&.each_with_index do |skipped_step, offset|
        skipped_idx = idx + 1 + offset
        step_results << {
          step: skipped_idx,
          description: skipped_step['description'].to_s,
          result: 'NOT EXECUTED — cost limit reached before this step could run.'
        }
        persist_plan_step_metadata!(
          step_index: skipped_idx,
          result_text: 'Skipped (cost cap)',
          status: 'skipped'
        )
      end
      break
    end

    chat_service_send(:emit_status, { event: :plan_step_complete, step_index: idx })
  end

  chat_service_send(:emit_status, { event: :plan_assembling })
  chat_service_send(:reset_full_response_for_plan!, +'')
  ass_in, ass_out, ass_cached = assemble_plan_response(
    goal: goal,
    step_results: step_results,
    streamer_proc: streamer_proc
  )
  total_in += ass_in
  total_out += ass_out
  total_cached += ass_cached

  @conversation.add_message(role: :assistant, content: chat_service_ivar(:@full_response))

  build_plan_execution_result(
    input_tokens: total_in,
    output_tokens: total_out,
    cached_tokens: total_cached,
    cache_creation_tokens: 0,
    tool_stats: merged_tool_stats,
    plan_tool_limit_reached: plan_tool_limit_reached
  )
rescue StandardError => e
  Rails.logger.error("[Assistant::ChatService] Plan orchestration crashed: #{e.class}: #{e.message}")
  Rails.logger.error(e.backtrace&.first(8)&.join("\n"))
  mark_remaining_plan_steps_failed!(@current_plan_step_index || 0, e.message)
  raise
end