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.

Constant Summary

DEFAULT_MAX_ITERS =

777

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 292

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 298

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.

      #{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 215

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

  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])
  messages = [
    { role: 'system', content: system_role_content },
    { role: 'user',   content: request }
  ]
  append_session(session_id: session_id, role: 'user', content: request)

  turn_fails = Hash.new(0)

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

    messages << msg
    calls = Array(msg[:tool_calls])

    if calls.empty?
      text = msg[:content].to_s
      append_session(session_id: session_id, role: 'assistant', content: text)
      Learning.auto_reflect(session_id: session_id, request: request, final: text) if defined?(Learning)
      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)
      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
        hint    = defined?(Mistakes) ? Mistakes.correction_hint(tool: name, error: tele[:err] || raw[0, 300]) : ''
        result  = guard_repeated_failure(
          name: name,
          count: [turn_fails[fkey], persist].max,
          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
  end

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