Module: PWN::AI::Agent::Learning

Defined in:
lib/pwn/ai/agent/learning.rb

Overview

PWN::AI::Agent::Learning is the self-improvement engine that closes the pwn-ai feedback loop. It captures task outcomes, mines session transcripts for durable lessons, promotes successful workflows into reusable skills, and prunes / consolidates persistent memory so the agent gets sharper over time instead of accumulating noise.

Data flows:

Loop.run --(tool telemetry)--> Metrics.record
Loop.run --(final answer)----> Learning.auto_introspect (opt-in)
model    --(tool calls)------> learning_note_outcome / _distill_skill
PromptBuilder <----------------- Learning.to_context + Metrics.to_context

Everything is file-backed under ~/.pwn so it survives across REPL restarts and is shared by every future session.

Constant Summary collapse

LEARNING_FILE = 
File.join(Dir.home, '.pwn', 'learning.jsonl')
MAX_MEMORY_ENTRIES = 
200
FAILURE_FINAL_RX = 
/\[pwn-ai\] (iteration budget exhausted|engine returned no message)|\b(i (was )?unable to|i could not|i couldn'?t|cannot proceed|failed to)\b/i

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



411
412
413
 
# File 'lib/pwn/ai/agent/learning.rb', line 411

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

.auto_introspect(opts = {}) ⇒ Object

Supported Method Parameters

PWN::AI::Agent::Learning.auto_introspect( session_id: 'required - id of the just-completed session', request: 'optional - original user request (for outcome logging)', final: 'optional - final assistant answer (for outcome logging)' )

Called by Loop.run when PWN::Env[:agent][:auto_introspect] is truthy. Never raises — learning must not break the primary loop.



205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
 
# File 'lib/pwn/ai/agent/learning.rb', line 205

public_class_method def self.auto_introspect(opts = {})
  session_id = opts[:session_id]
  return unless session_id
  return unless auto_introspect_enabled?

  ok = infer_success(session_id: session_id, final: opts[:final])
  note_outcome(
    task: opts[:request].to_s[0, 120],
    success: ok,
    details: opts[:final].to_s[0, 300],
    session_id: session_id,
    tags: %w[auto loop]
  )
  reflect(session_id: session_id)
  Extrospection.auto_extrospect(session_id: session_id) if defined?(Extrospection)
rescue StandardError => e
  warn "[pwn-ai/learning] auto_introspect swallowed: #{e.class}: #{e.message}"
  nil
end

.consolidate(opts = {}) ⇒ Object

Supported Method Parameters

removed = PWN::AI::Agent::Learning.consolidate( max_entries: 'optional - hard cap on PWN::Memory size (default MAX_MEMORY_ENTRIES)' )

Deduplicates near-identical lesson values and prunes the oldest entries once the cap is exceeded so the injected MEMORY block stays high-signal.



265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
 
# File 'lib/pwn/ai/agent/learning.rb', line 265

public_class_method def self.consolidate(opts = {})
  cap = opts[:max_entries] || MAX_MEMORY_ENTRIES
  return { removed: 0 } unless defined?(PWN::Memory)

  mem = PWN::Memory.load
  seen = {}
  removed = []
  mem.each do |k, v|
    sig = Digest::SHA256.hexdigest(v[:value].to_s.strip.downcase)[0, 16]
    if seen[sig]
      removed << k
    else
      seen[sig] = k
    end
  end
  removed.each { |k| mem.delete(k) }

  if mem.size > cap
    sorted = mem.sort_by { |_k, v| v[:timestamp].to_s }
    drop   = sorted.first(mem.size - cap).map(&:first)
    drop.each { |k| mem.delete(k) }
    removed.concat(drop)
  end
  PWN::Memory.save(mem: mem)
  { removed: removed.length, remaining: mem.size }
end

.distill_skill(opts = {}) ⇒ Object

Supported Method Parameters

skill = PWN::AI::Agent::Learning.distill_skill( name: 'required - snake_case name for the new skill', session_id: 'optional - PWN::Sessions id to mine (uses its transcript)', content: 'optional - explicit markdown body; overrides transcript mining', references: 'optional - Array of reference URLs / CWE / CVE / ATT&CK ids' )



138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/pwn/ai/agent/learning.rb', line 138

public_class_method def self.distill_skill(opts = {})
  name = opts[:name].to_s.gsub(/[^a-z0-9_-]/i, '_')
  raise 'ERROR: name is required' if name.empty?

  body = opts[:content].to_s
  body = build_skill_from_session(session_id: opts[:session_id], name: name) if body.strip.empty? && opts[:session_id]
  raise 'ERROR: content or session_id is required' if body.strip.empty?

  refs = Array(opts[:references]).map(&:to_s).map(&:strip).reject(&:empty?).uniq
  unless refs.empty?
    body = "---\nreferences:\n#{refs.map { |r| "  - #{r}" }.join("\n")}\n---\n#{body}" unless body.start_with?("---\n")
    body = "#{body.rstrip}\n\n## References\n#{refs.map { |r| "- #{r}" }.join("\n")}\n" unless body =~ /^\#{1,3}\s*References\s*$/i
  end

  dir = skills_dir
  FileUtils.mkdir_p(dir)
  path = File.join(dir, "#{name}.md")
  File.write(path, body)
  PWN::Config.load_skills(pwn_skills_path: dir) if defined?(PWN::Config) && PWN::Config.respond_to?(:load_skills)
  note_outcome(task: "distill_skill:#{name}", success: true, details: "Saved #{path}", tags: %w[skill auto])
  { saved: true, name: name, path: path, bytes: body.bytesize, references: refs }
end

.flip_last_outcome(opts = {}) ⇒ Object

Supported Method Parameters

PWN::AI::Agent::Learning.flip_last_outcome( session_id: 'optional - only flip if the newest outcome belongs to this session', reason: 'optional - why it is being flipped (usually the user correction text)' )

Rewrites the most-recently-appended learning.jsonl entry from success:true to success:false. Called by Mistakes.check_user_correction when the user's next message rejects the previous answer, so the 100 %-success illusion is broken and the failure enters the corpus.



236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/pwn/ai/agent/learning.rb', line 236

public_class_method def self.flip_last_outcome(opts = {})
  return { flipped: false } unless File.exist?(LEARNING_FILE)

  lines = File.readlines(LEARNING_FILE)
  return { flipped: false } if lines.empty?

  last = JSON.parse(lines.last, symbolize_names: true)
  return { flipped: false } if opts[:session_id] && last[:session_id] && last[:session_id] != opts[:session_id]
  return { flipped: false } unless last[:success]

  last[:success]    = false
  last[:flipped_by] = 'user_correction'
  last[:details]    = "#{last[:details]} | CORRECTED: #{opts[:reason].to_s[0, 200]}".strip
  lines[-1] = "#{JSON.generate(last)}\n"
  File.write(LEARNING_FILE, lines.join)
  { flipped: true, id: last[:id] }
rescue StandardError
  { flipped: false }
end

.help ⇒ Object

Display Usage for this Module



417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
 
# File 'lib/pwn/ai/agent/learning.rb', line 417

public_class_method def self.help
  puts <<~USAGE
    USAGE:
      PWN::AI::Agent::Learning.note_outcome(task: 'nmap sweep 10.0.0.0/24', success: true, details: '12 hosts up')
      PWN::AI::Agent::Learning.outcomes(limit: 20, success: false)
      PWN::AI::Agent::Learning.reflect(session_id: sid)              # LLM or heuristic → PWN::Memory
      PWN::AI::Agent::Learning.auto_introspect(session_id: sid, request: req, final: text)
      PWN::AI::Agent::Learning.distill_skill(name: 'quick_recon', session_id: sid)
      PWN::AI::Agent::Learning.consolidate(max_entries: 200)         # dedupe + prune Memory
      PWN::AI::Agent::Learning.to_context(limit: 5)                  # injected by PromptBuilder
      PWN::AI::Agent::Learning.stats
      PWN::AI::Agent::Learning.reset

      Enable end-of-run auto-learning with:
        PWN::Env[:ai][:agent][:auto_introspect] = true

      #{self}.authors
  USAGE
end

.note_outcome(opts = {}) ⇒ Object

Supported Method Parameters

entry = PWN::AI::Agent::Learning.note_outcome( task: 'required - short description of what was attempted', success: 'required - Boolean, did the attempt achieve its goal', details: 'optional - free-form notes / error / evidence', session_id: 'optional - PWN::Sessions id this outcome belongs to', tags: 'optional - Array of String labels for later retrieval' )



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/pwn/ai/agent/learning.rb', line 38

public_class_method def self.note_outcome(opts = {})
  task    = opts[:task].to_s
  success = opts[:success] ? true : false
  raise 'ERROR: task is required' if task.strip.empty?

  entry = {
    id: Digest::SHA256.hexdigest("#{task}-#{Time.now.to_f}")[0, 12],
    task: task,
    success: success,
    details: opts[:details].to_s[0, 2_000],
    session_id: opts[:session_id],
    tags: Array(opts[:tags]).map(&:to_s),
    timestamp: Time.now.utc.iso8601
  }
  FileUtils.mkdir_p(File.dirname(LEARNING_FILE))
  File.open(LEARNING_FILE, 'a') { |f| f.puts(JSON.generate(entry)) }

  key = :"lesson_#{entry[:id]}"
  cat = :lesson
  val = "#{success ? 'SUCCESS' : 'FAILURE'}: #{task} — #{opts[:details].to_s.strip[0, 200]}"
  PWN::Memory.remember(key: key, value: val, category: cat) if defined?(PWN::Memory)

  entry
end

.outcomes(opts = {}) ⇒ Object

Supported Method Parameters

rows = PWN::AI::Agent::Learning.outcomes( limit: 'optional - max entries returned newest-first (default 50)', success: 'optional - filter by Boolean outcome', tag: 'optional - filter by tag substring' )



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/pwn/ai/agent/learning.rb', line 70

public_class_method def self.outcomes(opts = {})
  limit   = opts[:limit] || 50
  want_ok = opts.key?(:success) ? !opts[:success].nil? && opts[:success] != false : nil
  tag     = opts[:tag].to_s.downcase
  return [] unless File.exist?(LEARNING_FILE)

  rows = File.readlines(LEARNING_FILE).map do |l|
    JSON.parse(l, symbolize_names: true)
  rescue StandardError
    nil
  end
  rows.compact!
  rows.select! { |r| r[:success] == want_ok } unless want_ok.nil?
  rows.select! { |r| Array(r[:tags]).any? { |t| t.to_s.downcase.include?(tag) } } unless tag.empty?
  rows.reverse.first(limit)
end

.reflect(opts = {}) ⇒ Object

Supported Method Parameters

report = PWN::AI::Agent::Learning.reflect( session_id: 'required - PWN::Sessions id to analyse', dry_run: 'optional - when true, do not write to Memory/Skills (default false)' )

Uses PWN::AI::Agent::Reflect (when available) to LLM-summarise the session into structured lessons. Falls back to a heuristic extractor when module_reflection is disabled so learning never stops.



171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/pwn/ai/agent/learning.rb', line 171

public_class_method def self.reflect(opts = {})
  session_id = opts[:session_id]
  dry_run    = opts[:dry_run] ? true : false
  raise 'ERROR: session_id is required' if session_id.to_s.empty?

  transcript = PWN::Sessions.load(session_id: session_id)
  return { session_id: session_id, lessons: [], reason: 'empty transcript' } if transcript.empty?

  lessons = introspective_lessons(transcript: transcript)
  lessons = heuristic_lessons(transcript: transcript) if lessons.empty?

  saved = []
  lessons.each do |l|
    next if l.to_s.strip.empty?

    key = :"reflect_#{session_id}_#{Digest::SHA256.hexdigest(l)[0, 8]}"
    PWN::Memory.remember(key: key, value: l, category: :lesson) unless dry_run
    saved << { key: key, lesson: l }
  end
  consolidate unless dry_run

  { session_id: session_id, lessons: saved, count: saved.length, dry_run: dry_run }
end

.reset ⇒ Object

Supported Method Parameters

PWN::AI::Agent::Learning.reset



295
296
297
298
 
# File 'lib/pwn/ai/agent/learning.rb', line 295

public_class_method def self.reset
  FileUtils.rm_f(LEARNING_FILE)
  { cleared: true }
end

.stats ⇒ Object

Supported Method Parameters

stats = PWN::AI::Agent::Learning.stats



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/pwn/ai/agent/learning.rb', line 90

public_class_method def self.stats
  rows   = outcomes(limit: 10_000)
  total  = rows.length
  ok     = rows.count { |r| r[:success] }
  skills = defined?(PWN::Skills) && PWN::Skills.is_a?(Hash) ? PWN::Skills.keys.length : 0
  mem    = defined?(PWN::Memory) ? PWN::Memory.load.keys.length : 0
  {
    total_outcomes: total,
    successes: ok,
    failures: total - ok,
    success_rate: total.positive? ? (ok.to_f / total).round(3) : 0.0,
    skills_known: skills,
    memory_entries: mem,
    tool_metrics: (Metrics.summary(limit: 5) if defined?(Metrics)),
    extrospection: (Extrospection.stats if defined?(Extrospection))
  }
end

.to_context(opts = {}) ⇒ Object

Supported Method Parameters

ctx = PWN::AI::Agent::Learning.to_context( limit: 'optional - number of recent outcomes to surface (default 5)' )



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/pwn/ai/agent/learning.rb', line 113

public_class_method def self.to_context(opts = {})
  limit = opts[:limit] || 5
  rows  = outcomes(limit: limit)
  fails = outcomes(limit: 200, success: false).first(limit)
  return '' if rows.empty? && fails.empty?

  fmt = lambda do |r|
    flag = r[:success] ? '✓' : '✗'
    "  #{flag} #{r[:task].to_s[0, 100]} (#{r[:timestamp]})"
  end
  s   = stats
  hdr = "RECENT OUTCOMES (success_rate=#{(s[:success_rate] * 100).round(1)}% over #{s[:total_outcomes]} attempts)"
  out = "#{hdr}\n#{rows.map(&fmt).join("\n")}\n"
  out += "RECENT FAILURES (learn from these — do not repeat)\n#{fails.map(&fmt).join("\n")}\n" unless fails.empty?
  "#{out}\n"
end