Module: PWN::AI::Agent::TaskSummarizer

Defined in:

lib/pwn/ai/agent/task_summarizer.rb

Overview

High-level executive brief of the work the agent is about to do.

English tangible tasks are PRIMARY (not tool jargon):

1. plan(request:) — on user submit, break the goal into an ordered
 list of plain-English tasks via the active LLM (each task is a
 coherent unit that may require many tool calls). Works for ANY
 request — no static per-domain task lists.
2. about_to(tools:) — per tool-batch brief led by the active
 "task k/n: <english>" item (same vocabulary as emit_plan!).
 Tool counts/intents are a secondary "via …" suffix only.
3. active_task_prompt / plan_context — injected into Loop messages
 so generated tasks steer tool choice, not only the TUI.
4. record! emits an advancement brief when plan_idx moves forward.

Never dumps raw commands or tool results into the task row — those stay on the per-tool lines the REPL already prints.

REPL on_tool contract (repl.rb):

on_tool.call('task', full_summary_text, '')  # result MUST be empty
# → [ ts → pwn-ai → task ] <full summary, no truncation>
on_tool.call('shell', args, result)          # real tool

Constant Summary

DEFAULT_EVERY =

5

DEFAULT_INTERVAL_S =

8.0

MAX_BUFFER =

64

PREVIEW_LEN =

Soft cap only for non-plan batch briefs when NOT showing full. Plan text and task summaries are always shown in full (no ellipsis).

2_000

MAX_PLAN_TASKS =

12

MIN_PLAN_TASKS =

2

ENGINE_MODS =

{
  openai: 'PWN::AI::OpenAI',
  grok: 'PWN::AI::Grok',
  ollama: 'PWN::AI::Ollama',
  anthropic: 'PWN::AI::Anthropic',
  gemini: 'PWN::AI::Gemini'
}.freeze

PLAN_SYSTEM =

<<~SYS
  You are the pwn-ai Task Planner. Given ANY user request, break it into
  an ordered list of tangible work units an autonomous security agent
  should perform. Rules:
  - 2..12 tasks. Each task is one coherent unit of work (may need many tools).
  - Plain US English. Imperative mood. No tool names, paths, or shell commands.
  - Cover discovery/recon, the core work, verification, and the requested
    deliverable/format when relevant.
  - Tailor steps to THIS request only — do not reuse a canned domain script.
  - Last task should verify or present the final result when that fits.
  - Output ONLY a JSON array of strings. No markdown, no prose, no keys.
  Example: ["determine the local IPv4 subnet","find live hosts on that subnet","present live hosts as JSON"]
SYS

TOOL_NAME_ROOTS =

Roots of registered agent tools — used to detect plan_first outlines that list tool calls instead of plain-English tangible work.

%w[
  shell pwn_eval
  memory sessions mistakes learning extro skill agent swarm cron
  reward curriculum metrics
].freeze

Class Method Summary

• .about_to(opts = {}) ⇒ Object

  High-level brief for a collection of impending tool calls.

• .active_task(opts = {}) ⇒ Object

  Active plain-English plan item (and index/n) for Loop / model steering.

• .active_task_prompt(opts = {}) ⇒ Object

  Inject / refresh the active-task focus into Loop messages when plan_idx changes.

• .apply_prm_advancement!(opts = {}) ⇒ Object

  Advance or hold plan_idx from an R2 step batch.

• .authors ⇒ Object

  Author(s)

  0day Inc.

• .chat_for_plan(opts = {}) ⇒ Object

  Prefer Reflect when module_reflection is on (teacher engine / gated).

• .emit!(opts = {}) ⇒ Object

  Optional progress / done line (verbose or flush).

• .emit_plan!(opts = {}) ⇒ Object

  Emit plan once at loop start.

• .enabled? ⇒ Boolean
• .every_n ⇒ Object
• .fallback_decompose(opts = {}) ⇒ Object

  Thin offline fallback when the LLM is disabled or unavailable.

• .flush!(opts = {}) ⇒ Object

  Supported Method Parameters

  line = PWN::AI::Agent::TaskSummarizer.flush!( state: 'required - fresh() hash' ).

• .format_plan(opts = {}) ⇒ Object

  Format the full plan as the task-summary body (shown in entirety).

• .fresh(opts = {}) ⇒ Object

  Per-run state (also safe for nested/swarm if callers keep their own hash) Supported Method Parameters

  state = PWN::AI::Agent::TaskSummarizer.fresh( request: 'optional - original user goal string' ).

• .help ⇒ Object

  Display Usage for this Module.

• .heuristic_decompose(opts = {}) ⇒ Object

  Back-compat alias used by older call sites / specs.

• .interval_s ⇒ Object
• .llm_plan_enabled? ⇒ Boolean

  LLM plan generation is on by default.

• .parse_llm_tasks(opts = {}) ⇒ Object

  Parse JSON array, fenced JSON, or numbered/bulleted plain text.

• .parse_outline_tasks(opts = {}) ⇒ Object

  Parse a plan_first / red_team surviving outline into tangible tasks.

• .plan(opts = {}) ⇒ Object

  ------------------------------------------------------------------ Request → ordered tangible tasks (each may map to many tools).

• .plan_context(opts = {}) ⇒ Object

  Short block for engine messages: full plan + focus on active English task.

• .record!(opts = {}) ⇒ Object

  Record a completed tool.

• .relevance_query(opts = {}) ⇒ Object

  Build a Registry/tool-router relevance string from English tasks.

• .tool_jargon_task?(opts = {}) ⇒ Boolean

  True when a candidate plan item is tool jargon (e.g. "shell", "shell / pwn_eval", "pwn_eval ×1") rather than a plain-English task.

• .unify_plan!(opts = {}) ⇒ Object

  After S4 red_team / plan_first: optionally rewrite ts_state from the surviving outline so the task line and adversarial plan are one object.

• .verbose? ⇒ Boolean

Class Method Details

.about_to(opts = {}) ⇒ Object

High-level brief for a collection of impending tool calls. English tangible task is PRIMARY; tool counts/intents are secondary. This string appears as name='task' and is shown in FULL in pwn-ai.

Supported Method Parameters

line = PWN::AI::Agent::TaskSummarizer.about_to( tools: 'optional - array of args: or bare names', name: 'optional - single tool name (legacy one-tool path)', args: 'optional - single tool args (ignored for brief content)', request: 'optional - goal text', state: 'optional - fresh() hash' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 662

public_class_method def self.about_to(opts = {})
  state   = opts[:state]
  request = state && state[:request].to_s
  request = opts[:request].to_s if request.nil? || request.empty?

  tools = normalize_tools(tools: opts[:tools], name: opts[:name], args: opts[:args])
  names = tools.map { |t| t[:name] }
  # Always ensure plan exists so mid-flight briefs can cite tangible tasks.
  plan(request: request, state: state) if state.is_a?(Hash) && Array(state[:plan]).empty? && !request.to_s.strip.empty?

  caps = capabilities_for(names: names)
  counts = tool_counts_phrase(names: names)
  intent = intent_phrase(tools: tools)

  via =
    if caps.empty?
      ''
    elsif intent != '' && counts != ''
      # Distinctive: tools + intent so shell/search ≠ shell/edit
      "via #{counts} (#{intent})"
    elsif counts != ''
      "via #{counts}"
    elsif caps.length == 1
      "via #{caps.first}"
    else
      head = caps[0..-2].join(', ')
      "via #{head}, and #{caps.last}"
    end

  # English task k/n is PRIMARY — same vocabulary as emit_plan!.
  # Previously tools led ("Next: shell×2 (search) [task k/n: …]") which
  # made mid-flight lines look like jargon and omitted the English task
  # on the first batch after the plan (operator "skipped tasks" complaint).
  plan_bit = ''
  has_plan = state.is_a?(Hash) && Array(state[:plan]).any?
  plan_emitted = state.is_a?(Hash) && state[:plan_emitted]
  if has_plan
    idx = active_plan_index(state: state)
    item = state[:plan][idx]
    if item
      state[:plan_idx] = idx if state.is_a?(Hash)
      plan_n = state[:plan].length
      # Always show English task k/n when a multi-step plan exists.
      # Single-task plans after emit_plan! already stated the only item —
      # keep via-only then to avoid a near-duplicate of the plan line.
      if plan_n <= 1 && plan_emitted
        plan_bit = ''
      else
        plan_bit = "task #{idx + 1}/#{plan_n}: #{item}"
      end
    end
  end

  # Goal lives on the emit_plan! line. Restate toward: only when this
  # brief would otherwise have no plan/goal linkage.
  why = why_bit(
    request: request,
    names: names,
    with_goal: plan_bit.empty? && !plan_emitted
  )

  # Compose: English task first, tools as via, optional why.
  line =
    if !plan_bit.empty? && !via.empty?
      "#{plan_bit} — #{via}"
    elsif !plan_bit.empty?
      plan_bit
    elsif !via.empty?
      "Next: #{via.sub(/\Avia /, '')}"
    else
      'Next: prepare the next step'
    end
  line = "#{line} — #{why}" unless why.empty?
  line = line.gsub(/[^\S\n]+/, ' ').strip
  # Full summary — never ellipsize. Pathological multi-MB blobs only
  # get a hard safety clamp far above normal executive briefs.
  line = line[0, 50_000] if line.length > 50_000

  # Suppress identical task lines when the model re-issues the same batch.
  if state.is_a?(Hash) && duplicate_brief?(state: state, line: line)
    state[:pending_tools] = names
    state[:emitted_for_batch] = true
    return nil
  end

  if state.is_a?(Hash)
    remember_brief!(state: state, line: line)
    state[:pending_tools] = names
    state[:emitted_for_batch] = true
  end
  line
rescue StandardError
  'Next: advance the current goal'
end

.active_task(opts = {}) ⇒ Object

Active plain-English plan item (and index/n) for Loop / model steering.

Supported Method Parameters

info = PWN::AI::Agent::TaskSummarizer.active_task( state: 'required - fresh() hash' ) => { idx:, n:, item:, label: "task k/n: …" } or nil

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 764

public_class_method def self.active_task(opts = {})
  state = opts[:state]
  return nil unless state.is_a?(Hash)

  plan = Array(state[:plan])
  return nil if plan.empty?

  idx = active_plan_index(state: state)
  item = plan[idx].to_s
  return nil if item.empty?

  n = plan.length
  {
    idx: idx,
    n: n,
    item: item,
    label: "task #{idx + 1}/#{n}: #{item}"
  }
rescue StandardError
  nil
end

.active_task_prompt(opts = {}) ⇒ Object

Inject / refresh the active-task focus into Loop messages when plan_idx changes. Returns the message content when a new injection is needed, else nil.

Supported Method Parameters

text = PWN::AI::Agent::TaskSummarizer.active_task_prompt( state: 'required - fresh() hash', force: 'optional - Boolean re-emit even if idx unchanged' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 851

public_class_method def self.active_task_prompt(opts = {})
  state = opts[:state]
  return nil unless state.is_a?(Hash)

  info = active_task(state: state)
  return nil unless info

  force = !opts[:force].nil?
  prev = state[:focus_injected_idx]
  return nil if !force && !prev.nil? && prev.to_i == info[:idx].to_i

  state[:focus_injected_idx] = info[:idx]
  # First injection after plan: full plan_context. Later: compact focus.
  if prev.nil? || force
    plan_context(state: state)
  else
    done_bit =
      if state[:last_advanced_from]
        from = state[:last_advanced_from].to_i
        prev_item = Array(state[:plan])[from]
        prev_item ? "Completed task #{from + 1}. " : ''
      else
        ''
      end
    "#{done_bit}[pwn-ai/tasks] Now focus on #{info[:label]}. " \
      'English tangible tasks solely drive tool choice — call only tools needed for THIS task; ' \
      'ignore PLAN: tool scaffolds and do not skip remaining tasks.'
  end
rescue StandardError
  nil
end

.apply_prm_advancement!(opts = {}) ⇒ Object

Advance or hold plan_idx from an R2 step batch. +1 streak matching the active task's tool intent -> advance once. Any -1 or mistake fingerprint on the batch -> do not advance.

Supported Method Parameters

idx = PWN::AI::Agent::TaskSummarizer.apply_prm_advancement!( state: 'required - fresh() hash', rewards: 'required - Array of -1|0|1 (batch order)', intents: 'optional - Array of intent verb strings for the batch', names: 'optional - tool names in the batch', mistake: 'optional - truthy when a mistake fingerprint hit this batch' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 1061

public_class_method def self.apply_prm_advancement!(opts = {})
  state = opts[:state]
  return nil unless state.is_a?(Hash)

  plan = Array(state[:plan])
  return state[:plan_idx].to_i if plan.length <= 1

  idx = state[:plan_idx].to_i
  return idx if idx >= plan.length - 1

  rewards = Array(opts[:rewards]).map(&:to_i)
  return idx if rewards.empty?

  # Hold on any regression or explicit mistake fingerprint.
  if opts[:mistake] || rewards.any?(&:negative?)
    state[:prm_pos_streak] = 0
    state[:last_prm_signal] = :hold_regress
    return idx
  end

  pos = rewards.count(&:positive?)
  neu = rewards.count(&:zero?)
  # Require a clear +1 presence (not all-neutral).
  if pos.zero?
    state[:prm_pos_streak] = 0
    state[:last_prm_signal] = :hold_neutral
    return idx
  end

  item = plan[idx].to_s.downcase
  intents = Array(opts[:intents]).map { |iv| iv.to_s.downcase }.reject(&:empty?)
  names = Array(opts[:names]).map(&:to_s)

  # When intents given, require at least one matches the active task language.
  matched =
    if intents.empty? && names.empty?
      true
    else
      intent_s = (intents + names).join(' ')
      task_intent_match?(item: item, intent: intent_s)
    end

  unless matched
    state[:last_prm_signal] = :hold_intent_mismatch
    return idx
  end

  streak = state[:prm_pos_streak].to_i + pos
  state[:prm_pos_streak] = streak
  # Advance after a streak of >=2 positive steps (or a single full +1 batch of size>=2).
  should = streak >= 2 || (pos >= 2 && neu.zero?)
  if should
    state[:last_advanced_from] = idx
    state[:plan_idx] = idx + 1
    state[:prm_pos_streak] = 0
    state[:tools_on_task] = 0
    state[:last_prm_signal] = :advance
    return idx + 1
  end

  state[:last_prm_signal] = :streak
  idx
rescue StandardError
  opts[:state].is_a?(Hash) ? opts[:state][:plan_idx].to_i : 0
end

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 1412

public_class_method def self.authors
  "AUTHOR(S):\n  0day Inc. <support@0dayinc.com>\n"
end

.chat_for_plan(opts = {}) ⇒ Object

Prefer Reflect when module_reflection is on (teacher engine / gated). Reflect.on uses direct engine .chat (never Loop.run) + depth guard, so this cannot re-enter via emit_plan! / after_read.

Supported Method Parameters

text = PWN::AI::Agent::TaskSummarizer.chat_for_plan( request: 'required - user goal to decompose' ) Kept public so specs can stub the LLM boundary.

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 440

public_class_method def self.chat_for_plan(opts = {})
  goal = opts[:request].to_s
  system = PLAN_SYSTEM
  user = "USER REQUEST:\n#{goal}\n\nReturn ONLY a JSON array of tangible task strings."
  if reflect_available?
    resp = Reflect.on(
      request: user,
      system_role_content: system,
      suppress_pii_warning: true,
      spinner: false
    )
    text = reflect_text(resp: resp)
    return text unless text.to_s.strip.empty?
  end

  # Fallback: active engine text chat (no tools). Used when reflection
  # is off, re-entrancy returned nil, or Reflect yielded empty.
  engine_chat(request: user, system_role_content: system)
rescue StandardError => e
  warn "[pwn-ai/task_summarizer] chat_for_plan swallowed: #{e.class}: #{e.message}"
  ''
end

.emit!(opts = {}) ⇒ Object

Optional progress / done line (verbose or flush). Still plain English; never includes raw tool results. Shown in full (no 60-char goal cut).

Supported Method Parameters

line = PWN::AI::Agent::TaskSummarizer.emit!( state: 'required - fresh() hash', final: 'optional - Boolean closing brief (default: false)' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 1360

public_class_method def self.emit!(opts = {})
  state = opts[:state]
  final = opts[:final]
  return nil if state.nil? || state[:events].empty?

  counts = state[:counts].sort_by { |_, c| -c }.map { |n, c| "#{n}×#{c}" }
  recent = state[:events].last([every_n, 1].max)
  caps = capabilities_for(names: recent.map { |e| e[:name] })
  focus =
    if caps.empty?
      'work'
    elsif caps.length == 1
      caps.first
    else
      "#{caps[0..-2].join(', ')}, and #{caps.last}"
    end
  fails = state[:events].count { |e| !e[:ok] }
  fail_bit = fails.positive? ? " (#{fails} hit issues)" : ''
  phase = final ? 'Finished' : 'Progress'
  goal = state[:request].to_s.gsub(/\s+/, ' ').strip
  goal_bit =
    if goal.empty?
      ''
    else
      # Full goal — no 60-char ellipsis.
      " toward: #{goal}"
    end
  plan_bit = ''
  if Array(state[:plan]).any?
    info = active_task(state: state)
    plan_bit =
      if info
        " | #{info[:label]}"
      else
        " | plan: #{state[:plan].length} tangible tasks"
      end
  end
  state[:since_emit] = 0
  state[:last_emit_at] = Time.now
  "#{phase}: #{focus} — #{state[:total]} tool calls so far (#{counts.first(6).join(', ')})#{fail_bit}#{goal_bit}#{plan_bit}"
end

.emit_plan!(opts = {}) ⇒ Object

Emit plan once at loop start. Returns full plan text (no truncation).

Supported Method Parameters

text = PWN::AI::Agent::TaskSummarizer.emit_plan!( state: 'required - fresh() hash', request: 'optional - goal override when state empty' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 364

public_class_method def self.emit_plan!(opts = {})
  state = opts[:state]
  return nil unless state.is_a?(Hash)
  return state[:plan_text] if state[:plan_emitted] && state[:plan_text]

  request = state[:request].to_s
  request = opts[:request].to_s if request.empty?
  tasks = state[:plan]
  tasks = plan(request: request, state: state) if tasks.nil? || Array(tasks).empty?
  text = format_plan(tasks: tasks, request: request)
  state[:plan] = Array(tasks)
  state[:plan_text] = text
  state[:plan_emitted] = true
  state[:plan_idx] = 0
  remember_brief!(state: state, line: text)
  text
rescue StandardError
  nil
end

.enabled? ⇒ Boolean

Returns:

• (Boolean)

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 61

public_class_method def self.enabled?
  v = PWN::Env.dig(:ai, :agent, :task_summary)
  v.nil? || !!v
rescue StandardError
  true
end

.every_n ⇒ Object

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 84

public_class_method def self.every_n
  n = PWN::Env.dig(:ai, :agent, :task_summary_every)
  n = DEFAULT_EVERY if n.nil?
  [n.to_i, 1].max
rescue StandardError
  DEFAULT_EVERY
end

.fallback_decompose(opts = {}) ⇒ Object

Thin offline fallback when the LLM is disabled or unavailable. Intentionally generic — NO static per-domain task scripts.

Supported Method Parameters

tasks = PWN::AI::Agent::TaskSummarizer.fallback_decompose( goal: 'required - user goal string' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 574

public_class_method def self.fallback_decompose(opts = {})
  goal_text = opts[:goal].to_s
  goal_lc = goal_text.downcase
  tasks = []

  # If the operator already spelled improvement bullets, surface them.
  bullets = goal_text.scan(/(?:^|\s)(?:\d+\.|[-*])\s*([^.;]+)/).flatten.map(&:strip)
  if bullets.length >= MIN_PLAN_TASKS
    bullets.first(8).each { |b| tasks << b.sub(/\Athe\s+/i, '').sub(/\.\s*\z/, '') }
    return tasks
  end

  tasks << "Understand the request: #{truncate_goal(goal: goal_text)}"
  tasks << "Carry out the core work for: #{truncate_goal(goal: goal_text)}"

  if goal_lc.match?(/\b(json|ya?ml|table|csv|tsv)\b/)
    fmt = goal_lc[/\b(json|ya?ml|table|csv|tsv)\b/]
    tasks << "Present the results in #{fmt} format"
  elsif goal_lc.match?(/\b(display|show|print|output|format|present|report|export)\b/)
    tasks << 'Present the final results in the requested format'
  end

  tasks << 'Run specs, rubocop, and/or rake to verify' if goal_lc.match?(/\b(test|spec|rubocop|rake|lint|verify|accept)\b/)

  tasks
rescue StandardError
  ["Carry out: #{truncate_goal(goal: opts[:goal])}"]
end

.flush!(opts = {}) ⇒ Object

Supported Method Parameters

line = PWN::AI::Agent::TaskSummarizer.flush!( state: 'required - fresh() hash' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 1406

public_class_method def self.flush!(opts = {})
  emit!(state: opts[:state], final: true)
end

.format_plan(opts = {}) ⇒ Object

Format the full plan as the task-summary body (shown in entirety). Uses "task k/n: ..." so emit_plan! and about_to share one vocabulary.

Supported Method Parameters

text = PWN::AI::Agent::TaskSummarizer.format_plan( tasks: 'required - Array of task strings', request: 'optional - goal string' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 338

public_class_method def self.format_plan(opts = {})
  list = Array(opts[:tasks]).map(&:to_s).reject(&:empty?)
  return '' if list.empty?

  goal = opts[:request].to_s.gsub(/\s+/, ' ').strip
  n = list.length
  lines = []
  lines << "Goal: #{goal}" unless goal.empty?
  lines << "Tangible tasks (#{n}) — each task may leverage one or more tools to complete its objective(s):"
  list.each_with_index do |t, i|
    lines << "  task #{i + 1}/#{n}: #{t}"
  end
  lines.join("\n")
rescue StandardError
  list = Array(opts[:tasks])
  n = list.length
  list.map.with_index(1) { |t, i| "task #{i}/#{n}: #{t}" }.join("\n")
end

.fresh(opts = {}) ⇒ Object

Per-run state (also safe for nested/swarm if callers keep their own hash)

Supported Method Parameters

state = PWN::AI::Agent::TaskSummarizer.fresh( request: 'optional - original user goal string' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 105

public_class_method def self.fresh(opts = {})
  {
    request: opts[:request].to_s,
    events: [],
    since_emit: 0,
    last_emit_at: Time.now,
    total: 0,
    counts: Hash.new(0),
    last_brief: nil,
    last_brief_fp: nil,
    pending_tools: [],
    emitted_for_batch: false,
    plan: [],
    plan_emitted: false,
    plan_text: nil,
    plan_idx: 0,
    batch_seq: 0,
    plan_source: nil,
    # RL-adjacent executive state (index only — credit lives in Reward)
    prm_pos_streak: 0,
    last_prm_signal: nil,
    unified_from: nil,
    # English-task-as-primary steering / advancement UX
    focus_injected_idx: nil,
    last_advanced_from: nil,
    last_advance_brief: nil,
    tools_on_task: 0
  }
end

.help ⇒ Object

Display Usage for this Module

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 1418

public_class_method def self.help
  puts <<~USAGE
    USAGE:
      state = PWN::AI::Agent::TaskSummarizer.fresh(request: 'ship task briefs to execs')
      # On user submit — LLM breaks ANY request into tangible tasks (full text):
      plan = PWN::AI::Agent::TaskSummarizer.plan(request: state[:request], state: state)
      text = PWN::AI::Agent::TaskSummarizer.emit_plan!(state: state)
      # → "Goal: ...\nTangible tasks (N) — each may use many tools:\n  task 1/N: ...\n  task 2/N: ..."
      # UI: on_tool.call('task', text, '')  # full text, no truncation
      # One task brief for a whole tool collection (one-to-many):
      pre = PWN::AI::Agent::TaskSummarizer.about_to(
        tools: [{ name: 'shell' }, { name: 'pwn_eval' }],
        state: state
      )
      # → "task k/N: <english> — via shell, pwn_eval (search+eval-ruby)"
      ctx = PWN::AI::Agent::TaskSummarizer.plan_context(state: state)
      focus = PWN::AI::Agent::TaskSummarizer.active_task_prompt(state: state)
      # then real tools print on their own lines; record! stays silent by default
      PWN::AI::Agent::TaskSummarizer.record!(
        state: state,
        name: 'shell',
        args: 'ls',
        result: '{success:true}'
      )
      line = PWN::AI::Agent::TaskSummarizer.flush!(state: state)  # optional closing brief
      PWN::AI::Agent::TaskSummarizer.enabled?
      PWN::AI::Agent::TaskSummarizer.verbose?
      PWN::AI::Agent::TaskSummarizer.llm_plan_enabled?
      PWN::AI::Agent::TaskSummarizer.unify_plan!(state: state, outline: plan_text)
      PWN::AI::Agent::TaskSummarizer.tool_jargon_task?(item: '`shell`')
      PWN::AI::Agent::TaskSummarizer.relevance_query(state: state, request: state[:request])
      PWN::AI::Agent::TaskSummarizer.apply_prm_advancement!(state: state, rewards: [1, 1], intents: ['search'])
      PWN::AI::Agent::TaskSummarizer.parse_outline_tasks(outline: plan_text)
      PWN::AI::Agent::TaskSummarizer.every_n
      PWN::AI::Agent::TaskSummarizer.interval_s

      #{self}.authors
  USAGE
end

.heuristic_decompose(opts = {}) ⇒ Object

Back-compat alias used by older call sites / specs.

Supported Method Parameters

tasks = PWN::AI::Agent::TaskSummarizer.heuristic_decompose( goal: 'required - user goal string' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 609

public_class_method def self.heuristic_decompose(opts = {})
  fallback_decompose(goal: opts[:goal])
end

.interval_s ⇒ Object

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 92

public_class_method def self.interval_s
  t = PWN::Env.dig(:ai, :agent, :task_summary_interval_s)
  t = DEFAULT_INTERVAL_S if t.nil?
  [t.to_f, 1.0].max
rescue StandardError
  DEFAULT_INTERVAL_S
end

.llm_plan_enabled? ⇒ Boolean

LLM plan generation is on by default. Set PWN::Env[:agent][:task_summary_llm] = false to force the offline generic fallback (tests / air-gapped).

Returns:

• (Boolean)

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 77

public_class_method def self.llm_plan_enabled?
  v = PWN::Env.dig(:ai, :agent, :task_summary_llm)
  v.nil? || !!v
rescue StandardError
  true
end

.parse_llm_tasks(opts = {}) ⇒ Object

Parse JSON array, fenced JSON, or numbered/bulleted plain text.

Supported Method Parameters

tasks = PWN::AI::Agent::TaskSummarizer.parse_llm_tasks( raw: 'required - raw LLM response text' ) Public so unit tests can exercise the parser without network I/O.

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 521

public_class_method def self.parse_llm_tasks(opts = {})
  text = opts[:raw].to_s.strip
  return [] if text.empty?

  # Strip thinking blocks some local models emit.
  text = text.gsub(%r{<think>.*?</think>}mi, '').strip
  text = text.sub(/\A```(?:json)?\s*/i, '').sub(/\s*```\z/, '').strip

  json_blob = text[/\[.*\]/m]
  if json_blob
    begin
      parsed = JSON.parse(json_blob)
      if parsed.is_a?(Array)
        tasks = parsed.map do |item|
          case item
          when String then item
          when Hash then (item['task'] || item[:task] || item['text'] || item[:text] || item.values.first).to_s
          else item.to_s
          end
        end
        cleaned = tasks.map { |t| t.to_s.gsub(/\s+/, ' ').strip }.reject(&:empty?)
        return cleaned if cleaned.length >= MIN_PLAN_TASKS
      end
    rescue JSON::ParserError
      # fall through to line parse
    end
  end

  lines = text.split(/\n+/).map(&:strip).reject(&:empty?)
  tasks = lines.filter_map do |ln|
    next if ln.match?(/\A[\[\]{},]\z/)
    next if ln.match?(/\A(?:here|tasks?|plan|json)\b/i) && ln.length < 40

    ln = ln.sub(/\A(?:\d+[.):]|[-*•])\s+/, '')
    ln = ln.sub(/\A["']/, '').sub(/["']\s*,?\s*\z/, '')
    ln = ln.sub(/,\s*\z/, '').strip
    next if ln.empty? || ln.length < 3
    next if ln.start_with?('[', '{')

    ln
  end
  tasks.uniq
rescue StandardError
  []
end

.parse_outline_tasks(opts = {}) ⇒ Object

Parse a plan_first / red_team surviving outline into tangible tasks. Numbered lines ("1. foo", "2) bar") preferred; bullet lines fallback.

Supported Method Parameters

tasks = PWN::AI::Agent::TaskSummarizer.parse_outline_tasks( outline: 'required - free-text plan outline' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 966

public_class_method def self.parse_outline_tasks(opts = {})
  text = opts[:outline].to_s
  return [] if text.strip.empty?

  tasks = []
  text.split(/\n+/).each do |ln|
    ln = ln.strip
    next if ln.empty?
    next if ln.match?(/\Ap\s*\(\s*success\s*\)\s*=/i)
    next if ln.match?(/\Aconfidence\s*=/i)
    next if ln.match?(/\APLAN:\s*\z/i)

    next unless (m = ln.match(/\A(?:\d+[.):]|[-*•])\s+(.+)\z/))

    item = m[1].to_s.strip
    # Strip trailing tool-arg noise common in plan_first
    item = item.sub(/\s+[—-]\s+.*\z/, '').strip
    tasks << item unless item.empty?
  end
  if tasks.length < MIN_PLAN_TASKS
    inline = text.scan(/(?:^|\s)\d+[.):]\s+([^\d]+?)(?=(?:\s+\d+[.):]\s+)|$)/)
    tasks = inline.flatten.map { |t| t.to_s.gsub(/\s+/, ' ').strip }.reject(&:empty?) if inline.length >= MIN_PLAN_TASKS
  end
  tasks = tasks.map { |t| t.to_s.gsub(/\s+/, ' ').strip }.reject(&:empty?).uniq
  tasks.first(MAX_PLAN_TASKS)
rescue StandardError
  []
end

.plan(opts = {}) ⇒ Object

---

Request → ordered tangible tasks (each may map to many tools). Called once when the user submits a request.

Priority:

1. Explicit numbered / bulleted steps already in the request
2. Active LLM decomposition (works for ANY request)
3. Thin generic offline fallback (never domain hardcoding)

Supported Method Parameters:: tasks = PWN::AI::Agent::TaskSummarizer.plan( request: 'required - user goal string', state: 'optional - fresh() hash to mutate', tasks: 'optional - injected plan array (tests)', llm_tasks: 'optional - injected LLM task array (tests)' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 284

public_class_method def self.plan(opts = {})
  goal = opts[:request].to_s.gsub(/\s+/, ' ').strip
  return [] if goal.empty?

  source = nil
  tasks = []

  # Prefer explicit enumerated steps already in the request (1. / 2. / - ).
  enumerated = extract_enumerated_steps(goal: goal)
  if enumerated.length >= MIN_PLAN_TASKS
    tasks = enumerated
    source = :enumerated
  elsif opts.key?(:tasks)
    # Caller-injected plan (tests / precomputed).
    tasks = Array(opts[:tasks]).map { |t| t.to_s.strip }.reject(&:empty?)
    source = :injected
  else
    tasks = llm_decompose(goal: goal, llm_tasks: opts[:llm_tasks], has_llm_tasks: opts.key?(:llm_tasks))
    source = tasks.any? ? :llm : nil
    if tasks.length < MIN_PLAN_TASKS
      tasks = fallback_decompose(goal: goal)
      source = :fallback
    end
  end

  tasks = normalize_task_list(tasks: tasks, goal: goal)
  if opts[:state].is_a?(Hash)
    opts[:state][:plan] = tasks
    opts[:state][:request] = goal if opts[:state][:request].to_s.empty?
    opts[:state][:plan_source] = source
  end
  tasks
rescue StandardError
  goal = opts[:request].to_s.gsub(/\s+/, ' ').strip
  goal.empty? ? [] : normalize_task_list(tasks: ["Carry out: #{goal}"], goal: goal)
end

.plan_context(opts = {}) ⇒ Object

Short block for engine messages: full plan + focus on active English task. Primary steering surface so tools follow generated tasks, not only TUI.

Supported Method Parameters

text = PWN::AI::Agent::TaskSummarizer.plan_context( state: 'required - fresh() hash' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 793

public_class_method def self.plan_context(opts = {})
  state = opts[:state]
  return nil unless state.is_a?(Hash)

  plan = Array(state[:plan]).map { |t| t.to_s.strip }.reject(&:empty?)
  return nil if plan.empty?

  info = active_task(state: state)
  n = plan.length
  lines = []
  lines << '[pwn-ai/tasks] English tangible tasks are the SOLE driver of which tools execute next.'
  lines << 'Work the ACTIVE task to completion (many tools ok), then advance.'
  lines << 'Do not skip ahead. Do not pick tools from the original request alone or from any PLAN: tool-call scaffold.'
  lines << 'Original goal stays in context; the English tasks below are the work breakdown.'
  lines << "Active: #{info[:label]}" if info
  lines << "Tangible tasks (#{n}):"
  plan.each_with_index do |t, i|
    marker = info && i == info[:idx] ? '▶' : ' '
    lines << "  #{marker} task #{i + 1}/#{n}: #{t}"
  end
  lines.join("\n")
rescue StandardError
  nil
end

.record!(opts = {}) ⇒ Object

Record a completed tool. Does NOT emit task lines with results. Returns a deferred progress brief only when every_n / interval fires AND verbose? is on; otherwise nil (silent coalesce).

Supported Method Parameters

line = PWN::AI::Agent::TaskSummarizer.record!( state: 'required - fresh() hash', name: 'required - tool name', args: 'optional - tool args', result: 'optional - tool result string' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 1292

public_class_method def self.record!(opts = {})
  state = opts[:state]
  name = opts[:name]
  args = opts[:args]
  result = opts[:result]
  return nil unless state

  preview = verbose? ? arg_snippet(args: args).to_s[0, 60] : ''
  rs = result.to_s
  ok = !rs.match?(/\A\s*\{?\s*"?(success|ok)"?\s*=>\s*false/i) &&
       !rs.match?(/ERROR:|Traceback|NoMethodError|StandardError/i)
  state[:events] << { name: name.to_s, preview: preview, ok: ok, t: Time.now }
  state[:events].shift while state[:events].size > MAX_BUFFER
  state[:counts][name.to_s] += 1
  state[:total] += 1
  state[:since_emit] += 1
  state[:tools_on_task] = state[:tools_on_task].to_i + 1
  state[:emitted_for_batch] = false
  intent = intent_phrase(tools: [{ name: name.to_s, args: args }])
  # Live R2-local signal from tool outcome (executive idx only).
  # Full ORM/PRM credit stays in Reward during auto_introspect.
  local_r =
    if ok then 1
    elsif rs.match?(/exit["\s:=]+1\b/i) && name.to_s == 'shell' then 0
    else -1
    end
  mistake_hit = rs.match?(%r{REPEATED FAILURE|KNOWN FIX|\[pwn-ai/mistakes\]}i)
  prev_idx = state[:plan_idx].to_i
  apply_prm_advancement!(
    state: state,
    rewards: [local_r],
    intents: [intent],
    names: [name.to_s],
    mistake: mistake_hit
  )
  # Heuristic phase-shift remains as a backstop when PRM streak has not fired.
  if state[:last_prm_signal] != :advance
    maybe_advance_plan!(
      state: state,
      names: [name.to_s],
      intent: intent
    )
  end

  # Clearer advancement UX: when plan_idx moves, emit English task k/n brief
  # so operators see the same vocabulary as emit_plan! / about_to.
  if state[:plan_idx].to_i > prev_idx
    brief = advancement_brief(state: state, from_idx: prev_idx)
    state[:last_advance_brief] = brief
    return brief if brief
  end

  # Default: no mid-flight task spam. Progress lines only when verbose.
  return nil unless verbose?

  due = state[:since_emit] >= every_n ||
        (Time.now - state[:last_emit_at]) >= interval_s
  due ? emit!(state: state) : nil
end

.relevance_query(opts = {}) ⇒ Object

Build a Registry/tool-router relevance string from English tasks. Prefer active task + full plan over the bare original request so generated tangible tasks drive which tools are exposed/ranked.

Supported Method Parameters

q = PWN::AI::Agent::TaskSummarizer.relevance_query( state: 'optional - fresh() hash', request: 'optional - original user goal fallback' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 827

public_class_method def self.relevance_query(opts = {})
  state = opts[:state]
  request = opts[:request].to_s
  request = state[:request].to_s if request.empty? && state.is_a?(Hash)
  parts = []
  if state.is_a?(Hash)
    info = active_task(state: state)
    parts << info[:item] if info && !info[:item].to_s.empty?
    Array(state[:plan]).each { |t| parts << t.to_s }
  end
  parts << request unless request.empty?
  parts.map { |p| p.to_s.gsub(/\s+/, ' ').strip }.reject(&:empty?).uniq.join(' ')
rescue StandardError
  opts[:request].to_s
end

.tool_jargon_task?(opts = {}) ⇒ Boolean

True when a candidate plan item is tool jargon (e.g. "shell", "shell / pwn_eval", "pwn_eval ×1") rather than a plain-English task. plan_first asks for "exact tool calls"; those must NOT replace the operator-facing English tangible task list.

Supported Method Parameters

yes = PWN::AI::Agent::TaskSummarizer.tool_jargon_task?( item: 'required - candidate task string' )

Returns:

• (Boolean)

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 928

public_class_method def self.tool_jargon_task?(opts = {})
  s = opts[:item].to_s.gsub(/\s+/, ' ').strip
  return false if s.empty?

  # Bare / backticked tool token(s), optional ×N counts, slashes.
  # e.g. "`shell`", "shell", "shell / pwn_eval", "pwn_eval ×1"
  if s.match?(%r{\A(?:`?[a-z][a-z0-9_]*`?(?:\s*[×xX]\s*\d+)?(?:\s*(?:/|,|→|->)\s*`?[a-z][a-z0-9_]*`?(?:\s*[×xX]\s*\d+)?)*)\z})
    token = s.downcase.gsub(%r{[`×x\d\s,→/>-]}, ' ').split.first.to_s
    return true if TOOL_NAME_ROOTS.include?(token) ||
                   TOOL_NAME_ROOTS.any? { |r| token.start_with?("#{r}_") }
  end

  # Leading tool name then args: "shell - rg foo", "pwn_eval: code"
  first = s.split(/[\s:—-]+/).first.to_s.downcase.gsub('`', '')
  return true if TOOL_NAME_ROOTS.include?(first) ||
                 TOOL_NAME_ROOTS.any? { |r| first.start_with?("#{r}_") }

  tokens = s.scan(/[A-Za-z][A-Za-z0-9_]{2,}/)
  return false if tokens.length > 8 # prose sentences stay English

  toolish = tokens.count do |t|
    tl = t.downcase
    TOOL_NAME_ROOTS.include?(tl) ||
      TOOL_NAME_ROOTS.any? { |r| tl.start_with?("#{r}_") }
  end
  # Majority tool tokens on a short line ⇒ jargon
  toolish.positive? && toolish >= (tokens.length + 1) / 2 && tokens.length <= 6
rescue StandardError
  false
end

.unify_plan!(opts = {}) ⇒ Object

After S4 red_team / plan_first: optionally rewrite ts_state from the surviving outline so the task line and adversarial plan are one object. Does not re-emit the plan line (TUI already showed the submit-time breakdown); later about_to batches use the unified idx. REFUSES tool-jargon outlines from plan_first ("exact tool calls") so English tangible tasks stay the sole operator + model vocabulary.

Supported Method Parameters

plan = PWN::AI::Agent::TaskSummarizer.unify_plan!( state: 'required - fresh() hash', outline: 'required - plan_first text and/or red_team hint', source: 'optional - :plan_first|:red_team|:merged (default :merged)' )

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 1008

public_class_method def self.unify_plan!(opts = {})
  state = opts[:state]
  return nil unless state.is_a?(Hash)

  outline = opts[:outline].to_s
  return Array(state[:plan]) if outline.strip.empty?

  tasks = parse_outline_tasks(outline: outline)
  return Array(state[:plan]) if tasks.length < MIN_PLAN_TASKS

  # plan_first prompts for "exact tool calls (name + key args)". Those
  # outlines must NOT clobber the English tangible plan from plan() /
  # emit_plan!. Operator TUI (plan_text) and model steering
  # (plan_context / about_to / active_task) must share one English list.
  jargon_n = tasks.count { |t| tool_jargon_task?(item: t) }
  if jargon_n >= (tasks.length + 1) / 2
    state[:unified_from] = outline.to_s[0, 500]
    state[:plan_source] = :kept_english if Array(state[:plan]).any?
    return Array(state[:plan])
  end

  # Drop any residual tool-jargon lines mixed into an English outline.
  tasks = tasks.reject { |t| tool_jargon_task?(item: t) }
  return Array(state[:plan]) if tasks.length < MIN_PLAN_TASKS

  # Keep a verify/close step when outline omitted one.
  tasks = normalize_task_list(tasks: tasks, goal: state[:request].to_s)
  prev_idx = state[:plan_idx].to_i
  state[:plan] = tasks
  state[:plan_source] = (opts[:source] || :merged).to_sym
  state[:unified_from] = outline.to_s[0, 500]
  # Preserve relative progress when possible; clamp into new length.
  state[:plan_idx] = prev_idx.clamp(0, [tasks.length - 1, 0].max)
  # Keep both UI surfaces in sync: refresh plan_text so the TUI
  # "Tangible tasks" block matches plan_context / about_to vocabulary.
  state[:plan_text] = format_plan(tasks: tasks, request: state[:request].to_s) if state[:plan_emitted] || state[:plan_text]
  tasks
rescue StandardError
  Array(opts[:state].is_a?(Hash) ? opts[:state][:plan] : nil)
end

.verbose? ⇒ Boolean

Returns:

• (Boolean)

# File 'lib/pwn/ai/agent/task_summarizer.rb', line 68

public_class_method def self.verbose?
  !!PWN::Env.dig(:ai, :agent, :task_summary_verbose)
rescue StandardError
  false
end