Module: PWN::AI::Agent::Loop

Defined in:

lib/pwn/ai/agent/loop.rb

Overview

The agent conversation loop:

build system prompt → call LLM with tools → if tool_calls: dispatch,
append role:'tool' results, loop → else: return text.

This replaces the regex-ReAct in PWN::Plugins::REPL :pwn_ai_hook with native function-calling. State (memory, skills, sessions) is all externalised — Loop.run is stateless aside from the messages array it builds.

NEGATIVE-FEEDBACK CLOSURE

Loop.run is where "learn from mistakes, don't repeat them" is actually enforced. On EVERY failed dispatch it:

1. Records the (tool, normalised_error) fingerprint into
 PWN::AI::Agent::Mistakes with a PERSISTENT cross-session count.
2. Reads that count back and, if it OR the in-turn count reaches
 REPEAT_THRESHOLD, prepends a hard "REPEATED FAILURE — change
 approach" guard to the tool result the model sees next.
3. Appends Mistakes.correction_hint (seen N×, sig, KNOWN FIX: …)
 so a previously-discovered fix is handed straight back to the
 model on the FIRST recurrence in a new session — it does not
 have to fail 3× again to re-learn what it already knew.

PromptBuilder.mistakes_block re-injects the top open mistakes and top known fixes into the system prompt of every future turn.

LOCAL-MODEL SCAFFOLDING

When the active engine is :ollama (or the corresponding :agent flags are set) Loop.run additionally:

* threads request → PromptBuilder for relevance-ranked MEMORY,
* threads request → Registry.definitions(relevance:) for a slimmed
tool set (:tool_router),
* splices Learning.exemplars_for(request:) between system and user
as few-shot behaviour retrieval,
* runs a plan-then-act pre-pass (:plan_first) so the model
externalises a tool plan before its first dispatch,
* escalates to a frontier persona for a 3-line corrective hint
once ≥ ESCALATE_AFTER_FAILS in-turn failures accumulate
(:escalation_persona) — the local model still produces the final
answer so Learning/Metrics stay attributed to :ollama.

Constant Summary

DEFAULT_MAX_ITERS =

777

ESCALATE_AFTER_FAILS =

4

BUDGET_HARD_STOP_FAILS =

P17 — when empty_final / known thrash shapes dominate, stop before burning the full ollama cap so the corpus is not pure terminal failure.

8

BUDGET_EMPTY_FINAL_STOP =

3

ENGINE_MODS =

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

Class Method Summary

• .authors ⇒ Object

  Author(s)

  0day Inc.

• .help ⇒ Object

  Display Usage for this Module.

• .run(opts = {}) ⇒ Object

  Supported Method Parameters

  final = PWN::AI::Agent::Loop.run( request: 'required - what the human typed', session_id: 'optional - PWN::Sessions id (transcript is appended to it)', enabled_toolsets: 'optional - subset of Registry.toolsets, or nil for all', on_tool: 'optional - ->(name, args, result) callback for live UI', system_role_content: 'optional - override default system prompt (built from session_id if not provided)' ).

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com

# File 'lib/pwn/ai/agent/loop.rb', line 679

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

.help ⇒ Object

Display Usage for this Module

# File 'lib/pwn/ai/agent/loop.rb', line 685

public_class_method def self.help
  puts <<~USAGE
    USAGE:
      final = PWN::AI::Agent::Loop.run(
        request: 'what does `id` return on this host?',
        session_id: PWN::Sessions.create[:id],
        enabled_toolsets: %w[terminal pwn memory skills],
        on_tool: ->(name, args, result) { puts "→ \#{name}: \#{result[0,1_024]}" },
        system_role_content: 'You are a helpful assistant that can call tools to answer questions.'
      )

      Supported engines: #{ENGINE_MODS.keys.join(', ')}
      Set PWN::Env[:ai][:active] to choose; PWN::Env[:ai][:agent][:max_iters] to bound.

      Local-model scaffolding (PWN::Env[:ai][:agent]):
        :plan_first          - Boolean, plan-then-act pre-pass (default: engine == :ollama)
        :tool_router         - Boolean/nil, slim Registry.definitions (nil=auto on for ollama)
        :escalation_persona  - Swarm persona name for frontier corrective hints when stuck
        :critic              - S3 constitutional critic before every final (Boolean)
        :red_team_plan       - S4 adversarial plan review after plan_first (Boolean)
        :counterfactual      - S2 A/B branch on REPEAT_THRESHOLD → DPO pair (Boolean)
        :hindsight           - C3 HER-relabel failures (Boolean, default true)
        :verify_as_reward    - E3 ground every final via extro_verify (Boolean)

      #{self}.authors
  USAGE
end

.run(opts = {}) ⇒ Object

Supported Method Parameters

final = PWN::AI::Agent::Loop.run( request: 'required - what the human typed', session_id: 'optional - PWN::Sessions id (transcript is appended to it)', enabled_toolsets: 'optional - subset of Registry.toolsets, or nil for all', on_tool: 'optional - ->(name, args, result) callback for live UI', system_role_content: 'optional - override default system prompt (built from session_id if not provided)' )

# File 'lib/pwn/ai/agent/loop.rb', line 517

public_class_method def self.run(opts = {})
  request = opts[:request].to_s
  session_id = opts[:session_id]
  on_tool = opts[:on_tool]
  engine = active_engine
  local  = engine == :ollama
  system_role_content = opts[:system_role_content] ||= PWN::AI::Agent::PromptBuilder.build(session_id: session_id, request: request)

  Registry.discover
  expose_current_session(session_id: session_id)
  Mistakes.check_user_correction(request: request, session_id: session_id) if defined?(Mistakes)

  tools    = Registry.definitions(enabled: opts[:enabled_toolsets], relevance: request)
  messages = [{ role: 'system', content: system_role_content }]
  messages.concat(Learning.exemplars_for(request: request)) if local && defined?(Learning) && Learning.respond_to?(:exemplars_for)
  messages << { role: 'user', content: request }
  append_session(session_id: session_id, role: 'user', content: request)

  predicted = nil
  Thread.current[:pwn_plan_predicted] = nil
  cal_state = calibration_state
  force_plan = cal_state[:force_plan]
  if (force_plan || agent_flag(key: :plan_first, default: local) || budget_exhaustion_hot?) && !Array(tools).empty?
    predicted = plan_first(messages: messages, request: request)
    # P22 — prefer explicit return; fall back to thread stash
    predicted = Thread.current[:pwn_plan_predicted] if predicted.nil?
  end
  if budget_exhaustion_hot?
    messages << {
      role: 'user',
      content: '[pwn-ai/p17] Budget-exhaustion is the top open failure on this host. Prefer the SHORTEST plan that finishes the ask (≤3 tool calls). Emit a final answer as soon as you have evidence — do not explore.'
    }
  end
  if force_plan && cal_state[:cal]
    messages << {
      role: 'user',
      content: "[pwn-ai/w3] engine=#{active_engine} is overconfident " \
               "(brier=#{cal_state[:cal][:brier]}, overconf=#{cal_state[:cal][:overconfidence]}). " \
               'Prefer high-judge exemplars, verify claims, and avoid speculative tool calls.'
    }
  end

  turn_fails = Hash.new(0)
  escalated  = false

  max_iters.times do |i|
    # 3.1 — compact history on local so tool dumps don't fill num_ctx
    compact_history!(messages: messages) if local

    msg = call_engine(messages: messages, tools: tools)
    return '[pwn-ai] engine returned no message' if msg.nil?

    calls = Array(msg[:tool_calls])
    text  = msg[:content].to_s

    # Empty-final guard (local/thinking models): Ollama sometimes
    # returns done_reason=stop with eval_count<=1, empty content, no
    # tool_calls — historically surface as a blank TUI reply. Do NOT
    # commit that as the answer; drop the empty assistant turn,
    # inject a one-shot nudge, and keep iterating.
    if calls.empty? && text.strip.empty?
      warn "[pwn-ai/loop] empty final from #{engine} on iter=#{i}; nudging" if local
      messages << {
        role: 'user',
        content: 'Your previous reply was empty (no tool_calls and no content). ' \
                 'Either call a tool now, or write the final answer for the user as plain text. ' \
                 'Do not reply with an empty message.'
      }
      turn_fails['empty_final'] += 1
      next
    end

    messages << msg

    if calls.empty?
      append_session(session_id: session_id, role: 'assistant', content: text)
      Learning.auto_introspect(session_id: session_id, request: request, final: text, predicted: predicted) if defined?(Learning) && should_auto_introspect?(local: local, turn_fails: turn_fails, iter: i)
      return text
    end

    calls.each do |tc|
      name    = tc.dig(:function, :name).to_s
      args    = tc.dig(:function, :arguments)
      entry   = Registry.lookup(name: name)
      started = Time.now
      raw     = Dispatch.call(tool_call: tc)
      tele    = record_metrics(name: name, started: started, raw: raw, args: args, session_id: session_id, engine: engine)
      result  = Result.condition(content: raw, entry: entry)

      unless tele[:ok]
        fkey = Digest::SHA256.hexdigest("#{name}|#{args}")[0, 16]
        turn_fails[fkey] += 1
        persist = tele.dig(:mistake, :count).to_i
        count   = [turn_fails[fkey], persist].max
        hint    = defined?(Mistakes) ? Mistakes.correction_hint(tool: name, error: tele[:err] || raw[0, 300]) : ''
        # S2 — counterfactual A/B: at the repeat threshold, fork an
        # alt-persona branch, judge both, inject the winner. Real
        # advantage estimation; (loser, winner) → DPO preference.
        thresh = defined?(Mistakes) ? Mistakes::REPEAT_THRESHOLD : 3
        if count >= thresh && !escalated && defined?(Curriculum)
          cf = (turn_fails["cf:#{fkey}"] += 1) == 1 ? Curriculum.counterfactual(request: request, name: name, args: args, error: tele[:err] || raw[0, 200], hint: hint) : nil
          hint = "#{hint}\n[pwn-ai/counterfactual] branch #{cf[:branch]} (score=#{cf[:score].round(2)}): #{cf[:content]}" if cf
        end
        result = guard_repeated_failure(name: name, count: count, hint: hint, result: result)
      end

      on_tool&.call(name, args, result)

      messages << {
        role: 'tool',
        tool_call_id: tc[:id] || tc['id'] || "call_#{i}",
        name: name,
        content: result
      }
      append_session(
        session_id: session_id,
        role: 'tool',
        content: "#{name} → #{result[0, 1_024]}"
      )
    end

    # P17 — hard stop: empty-final thrash or cumulative fails past cap.
    # Prefer a short apologetic final over another 10 useless tool dumps
    # that poison ORM/PRM/DPO with terminal failures.
    empty_n = turn_fails['empty_final'].to_i
    fail_n  = turn_fails.values.sum
    if empty_n >= BUDGET_EMPTY_FINAL_STOP || fail_n >= BUDGET_HARD_STOP_FAILS
      msg = if empty_n >= BUDGET_EMPTY_FINAL_STOP
              '[pwn-ai] stopped: repeated empty finals (budget thrash guard)'
            else
              '[pwn-ai] stopped: too many in-turn failures (budget thrash guard)'
            end
      if defined?(Mistakes)
        Mistakes.record(
          tool: 'agent_loop',
          error: "budget thrash guard fired empty=#{empty_n} fails=#{fail_n} iter=#{i}",
          session_id: session_id,
          source: :loop,
          shape: :budget_thrash
        )
      end
      append_session(session_id: session_id, role: 'assistant', content: msg)
      Learning.auto_introspect(session_id: session_id, request: request, final: msg, predicted: predicted) if defined?(Learning) && should_auto_introspect?(local: local, turn_fails: turn_fails, iter: i)
      return msg
    end

    next unless local && !escalated && turn_fails.values.sum >= ESCALATE_AFTER_FAILS

    hint = escalate(request: request, turn_fails: turn_fails, session_id: session_id)
    if hint
      messages << { role: 'tool', tool_call_id: "escalation_#{i}", name: 'frontier_hint', content: hint }
      append_session(session_id: session_id, role: 'tool', content: "frontier_hint → #{hint[0, 1_024]}")
    end
    escalated = true
  end

  Mistakes.record(tool: 'agent_loop', error: 'iteration budget exhausted without a final answer', session_id: session_id, source: :loop, shape: :budget_exhausted) if defined?(Mistakes)
  '[pwn-ai] iteration budget exhausted'
end