Module: PWN::AI::Agent::Curriculum

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

Overview

PWN::AI::Agent::Curriculum is Tier 4/5 of the pwn-ai reinforcement loop — the SELF-PLAY layer that turns the agent from a passive experience-recorder into an active learner:

S1  .practice        — Mistake-driven auto-curriculum. Reads
                     Mistakes.top(unresolved), asks Reflect to
                     generate 3 minimal reproducer prompts per
                     signature, self-plays each under Loop.run,
                     and auto-mistakes_resolve when Reward.judge
                     says the practice run solved it. THE AGENT
                     PRACTISES ITS OWN WEAKNESSES OVERNIGHT.
S2  .counterfactual  — On a repeated in-turn failure, forks: branch
                     A continues with the correction_hint, branch
                     B asks an alt persona for a different tool.
                     Reward.judge picks the winner; (loser,
                     winner) → Reward.record_preference. Real
                     advantage estimation, not imagined rollouts.
S3  .critic          — Constitutional critic persona with TOOL
                     ACCESS (can shell/extro_verify the claim).
                     Runs BEFORE note_outcome; its verdict feeds
                     Reward.judge and its concrete flaw becomes a
                     preference pair when the agent self-corrects.
S4  .red_team_plan   — After plan_first, an adversarial persona
                     reviews the plan against THIS host's
                     Metrics/Mistakes/extro_drift and injects a
                     pre-emptive correction_hint on the step it
                     predicts will fail.
C3  .hindsight       — Hindsight Experience Replay. On failure,
                     asks the judge "what DID this trajectory
                     accomplish?", relabels the episode with the
                     achieved-goal as success:true. Free positive
                     samples from failures — first HER on real
                     tool traces.
W2  .train_and_gate  — export_finetune + export_dpo → local LoRA
                     (unsloth/axolotl if installed) → replay
                     Mistakes.top on vN vs vN+1 → promote iff
                     resolved(N+1) > resolved(N). Fully autonomous
                     weight-level self-improvement with a
                     regression gate.
W3  .calibrate       — Tracks plan_first predicted p(success) vs
                     actual outcome → Brier score in Metrics.

All entry points are cron-safe (never raise into the caller) and depth-guarded via Swarm's Thread.current so a curriculum run cannot recurse into itself.

Constant Summary collapse

CURRICULUM_DIR = 
File.join(Dir.home, '.pwn', 'curriculum')
MODELS_FILE = 
File.join(CURRICULUM_DIR, 'models.json')
CRITIC_NAME = 
'pwn_critic'
RED_TEAM_NAME = 
'pwn_red_team'
ALT_NAME = 
'pwn_alt'
DPO_DIR_CONST = 
File.join(Dir.home, '.pwn', 'finetune')

Class Method Summary collapse

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com



523
524
525
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 523

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

.calibrate(opts = {}) ⇒ Object

Supported Method Parameters

PWN::AI::Agent::Curriculum.calibrate(predicted:, actual:, engine:)



300
301
302
303
304
305
306
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 300

public_class_method def self.calibrate(opts = {})
  p = opts[:predicted].to_f.clamp(0.0, 1.0)
  a = opts[:actual].to_f.clamp(0.0, 1.0)
  brier = (p - a)**2
  Metrics.record_calibration(predicted: p, actual: a, brier: brier, engine: opts[:engine]) if defined?(Metrics) && Metrics.respond_to?(:record_calibration)
  { predicted: p, actual: a, brier: brier.round(4) }
end

.counterfactual(opts = {}) ⇒ Object

Supported Method Parameters

winner = PWN::AI::Agent::Curriculum.counterfactual( request: 'required - original user request', name: 'required - tool that keeps failing', args: 'required - args it was called with', error: 'required - the failure text', hint: 'optional - branch-A correction_hint (from Mistakes)' )

Returns { branch: :a|:b, content:, score: } — the winning branch's suggestion, ready for Loop.run to inject as a synthetic tool result. Loser/winner is written to Reward.preferences (W1).



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 120

public_class_method def self.counterfactual(opts = {})
  return nil unless enabled?(key: :counterfactual)
  return nil if in_curriculum?

  request = opts[:request].to_s
  ensure_persona(name: ALT_NAME, role: 'You are an alternative-approach generator for pwn-ai. Given a failing tool call, propose ONE concrete DIFFERENT tool + args that would achieve the same sub-goal on this host. Reply with the tool call only, no prose.')

  branch_a = opts[:hint].to_s.strip
  branch_a = "retry #{opts[:name]} with corrected args" if branch_a.empty?
  branch_b = with_curriculum_guard do
    ask_persona(name: ALT_NAME, request: "Goal: #{request[0, 300]}\nFailing: #{opts[:name]}(#{opts[:args].to_s[0, 200]}) → #{opts[:error].to_s[0, 200]}\nPropose ONE different tool+args.")
  end
  return nil if branch_b.to_s.strip.empty?

  sa = score_branch(request: request, branch: branch_a)
  sb = score_branch(request: request, branch: branch_b)
  winner, loser, tag = sb > sa ? [branch_b, branch_a, :b] : [branch_a, branch_b, :a]
  Reward.record_preference(prompt: "#{request} | failing: #{opts[:name]} → #{opts[:error]}", rejected: loser, chosen: winner, source: :counterfactual) if defined?(Reward)
  { branch: tag, content: winner, score: [sa, sb].max, a: sa, b: sb }
rescue StandardError => e
  warn "[pwn-ai/curriculum] counterfactual swallowed: #{e.class}: #{e.message}"
  nil
end

.critic(opts = {}) ⇒ Object

Supported Method Parameters

v = PWN::AI::Agent::Curriculum.critic( request: 'required - user request', final: 'required - candidate final answer', session_id: 'optional - for evidence lookup' )

Returns { verdict: :pass|:flaw, flaw:, confidence: }. On :flaw the (final, flaw) pair is recorded as a preference so a future self-correction becomes DPO signal.



159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 159

public_class_method def self.critic(opts = {})
  return { verdict: :pass, source: :disabled } unless enabled?(key: :critic)
  return { verdict: :pass, source: :recursion } if in_curriculum?

  ensure_persona(name: CRITIC_NAME, role: "You are pwn-ai's constitutional critic. Given a REQUEST and a candidate ANSWER, find ONE concrete, verifiable flaw (wrong fact, missing step, unsupported claim, broken command). You MAY call shell / extro_verify / pwn_eval to check. If none found reply exactly: PASS. Otherwise reply: FLAW: <one line>.")
  reply = with_curriculum_guard do
    ask_persona(name: CRITIC_NAME, request: "REQUEST:\n#{opts[:request].to_s[0, 800]}\n\nANSWER:\n#{opts[:final].to_s[0, 2_000]}")
  end
  if reply.to_s.strip.upcase.start_with?('PASS')
    { verdict: :pass, confidence: 0.7 }
  else
    flaw = reply.to_s.sub(/\AFLAW:\s*/i, '').strip[0, 300]
    Mistakes.record(tool: 'assistant_answer', error: "critic: #{flaw}", args: opts[:final].to_s[0, 200], session_id: opts[:session_id], source: :model) if defined?(Mistakes)
    { verdict: :flaw, flaw: flaw, confidence: 0.7 }
  end
rescue StandardError => e
  { verdict: :pass, error: e.message }
end

.help ⇒ Object

Display Usage for this Module



529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 529

public_class_method def self.help
  puts <<~USAGE
    USAGE:
      # Tier 4 — self-play
      PWN::AI::Agent::Curriculum.practice(limit: 3)                     # S1 mistake-driven auto-curriculum
      PWN::AI::Agent::Curriculum.counterfactual(request:, name:, args:, error:, hint:)  # S2 A/B → DPO pair
      PWN::AI::Agent::Curriculum.critic(request:, final:)               # S3 tool-armed constitutional critic
      PWN::AI::Agent::Curriculum.red_team_plan(request:, plan:)         # S4 telemetry-grounded plan review
      PWN::AI::Agent::Curriculum.hindsight(request:, final:, session_id:) # C3 HER relabel

      # Tier 5 — close the weight loop
      PWN::AI::Agent::Curriculum.train_and_gate(dry_run: true)          # W2 SFT+DPO→LoRA→A/B gate→promote
      PWN::AI::Agent::Curriculum.calibrate(predicted: 0.8, actual: 1.0) # W3 Brier → Metrics[:calibration]

      Cron self-improvement (nightly):
        PWN::Cron.create(name: 'self_play', schedule: '0 3 * * *',
          ruby: 'PWN::AI::Agent::Curriculum.practice(limit: 5)')
        PWN::Cron.create(name: 'weight_loop', schedule: '0 4 * * 0',
          ruby: 'PWN::AI::Agent::Curriculum.train_and_gate(dry_run: false)')

      Config (PWN::Env[:ai][:agent]):
        :critic         - Boolean, run S3 before every note_outcome
        :red_team_plan  - Boolean, run S4 after every plan_first
        :counterfactual - Boolean, run S2 on REPEAT_THRESHOLD
        :hindsight      - Boolean, HER-relabel failures (default true)

      #{self}.authors
  USAGE
end

.hindsight(opts = {}) ⇒ Object

Supported Method Parameters

PWN::AI::Agent::Curriculum.hindsight( request: 'required - the FAILED goal', final: 'required - the final produced anyway', session_id: 'required - trajectory to relabel' )



216
217
218
219
220
221
222
223
224
225
226
227
228
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 216

public_class_method def self.hindsight(opts = {})
  return nil unless enabled?(key: :hindsight, default: true)
  return nil unless reflect_available?

  req = "The agent FAILED at: #{opts[:request].to_s[0, 300]}\nBut it produced: #{opts[:final].to_s[0, 800]}\n\nIn ≤12 words, what goal DID this trajectory accomplish? Reply with the goal only, or NOTHING if truly nothing."
  achieved = Reflect.on(request: req, suppress_pii_warning: true).to_s.strip
  return nil if achieved.empty? || achieved.upcase == 'NOTHING' || achieved.length > 200

  Learning.note_outcome(task: achieved, success: true, details: "HER-relabelled from failed: #{opts[:request].to_s[0, 100]}", session_id: opts[:session_id], tags: %w[hindsight her], score: 0.7) if defined?(Learning)
  { original: opts[:request].to_s[0, 100], achieved: achieved }
rescue StandardError
  nil
end

.practice(opts = {}) ⇒ Object

Supported Method Parameters

report = PWN::AI::Agent::Curriculum.practice( limit: 'optional - top-N unresolved mistakes to practise (default 3)', prompts_per: 'optional - reproducer prompts per mistake (default 2)', dry_run: 'optional - generate prompts but do not self-play (default false)' )



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
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 74

public_class_method def self.practice(opts = {})
  limit   = (opts[:limit] || 3).to_i
  per     = (opts[:prompts_per] || 2).to_i
  dry_run = opts[:dry_run] ? true : false
  return { skipped: 'recursion guard' } if in_curriculum?

  FileUtils.mkdir_p(CURRICULUM_DIR)
  targets = defined?(Mistakes) ? Mistakes.top(limit: limit, unresolved_only: true) : []
  results = []

  with_curriculum_guard do
    targets.each do |m|
      prompts = generate_reproducers(mistake: m, count: per)
      runs = dry_run ? [] : prompts.map { |p| self_play(prompt: p, tag: "practice:#{m[:signature]}") }
      solved = runs.select { |r| r[:score].to_f >= 0.7 }
      if !solved.empty? && defined?(Mistakes)
        fix = solved.max_by { |r| r[:score] }[:final].to_s.lines.first(3).join.strip[0, 400]
        Mistakes.resolve(signature: m[:signature], fix: "auto-curriculum: #{fix}")
        Reward.record_preference(prompt: prompts.first, rejected: m[:snippet].to_s, chosen: fix, source: :curriculum) if defined?(Reward)
      end
      results << { signature: m[:signature], tool: m[:tool], prompts: prompts, runs: runs.map { |r| { score: r[:score], verdict: r[:verdict] } }, resolved: !solved.empty? }
    end
  end
  log(event: :practice, data: results)
  { practiced: results.length, resolved: results.count { |r| r[:resolved] }, results: results, dry_run: dry_run }
rescue StandardError => e
  { error: "#{e.class}: #{e.message}" }
end

.red_team_plan(opts = {}) ⇒ Object

Supported Method Parameters

hint = PWN::AI::Agent::Curriculum.red_team_plan( request: 'required - user goal', plan: 'required - numbered plan text from plan_first' )



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 188

public_class_method def self.red_team_plan(opts = {})
  return nil unless enabled?(key: :red_team_plan)
  return nil if in_curriculum?

  ensure_persona(name: RED_TEAM_NAME, role: 'You are pwn-ai\'s adversarial plan reviewer. Given a numbered tool plan and telemetry from THIS host (tool success rates, known mistakes, environment drift), identify the ONE step most likely to fail and say why in ≤2 lines. Cite the metric/mistake/drift. If the plan is sound reply: SOUND.')
  telemetry = build_telemetry
  reply = with_curriculum_guard do
    ask_persona(name: RED_TEAM_NAME, request: "GOAL: #{opts[:request].to_s[0, 300]}\n\nPLAN:\n#{opts[:plan].to_s[0, 1_200]}\n\nHOST TELEMETRY:\n#{telemetry}")
  end
  return nil if reply.to_s.strip.upcase.start_with?('SOUND') || reply.to_s.strip.empty?

  "[pwn-ai/red_team] pre-emptive: #{reply.to_s.strip[0, 400]}"
rescue StandardError => e
  warn "[pwn-ai/curriculum] red_team_plan swallowed: #{e.class}: #{e.message}"
  nil
end

.train_and_gate(opts = {}) ⇒ Object

Supported Method Parameters

r = PWN::AI::Agent::Curriculum.train_and_gate( base_model: 'optional - ollama base tag (default PWN::Env[:ollama][:model])', trainer: 'optional - :unsloth | :axolotl | :auto (default :auto)', dry_run: 'optional - export + build eval set but do not train (default true)' )

Best-effort orchestrator. When a supported trainer is installed it produces ~/.pwn/finetune/pwn-vN/, cuts an ollama Modelfile with the LoRA adapter, then REPLAYS Mistakes.top against vN and vN+1 under Reward.judge. Promotes (writes MODELS_FILE) iff vN+1 resolves more signatures. When no trainer is present it still exports SFT+DPO and emits the exact CLI to run manually — so the pipeline is complete even on a box without GPU.



249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
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
291
 
# File 'lib/pwn/ai/agent/curriculum.rb', line 249

public_class_method def self.train_and_gate(opts = {})
  dry_run = opts.key?(:dry_run) ? opts[:dry_run] : true
  FileUtils.mkdir_p(CURRICULUM_DIR)
  sft = defined?(Learning) ? Learning.export_finetune(format: :sharegpt) : nil
  dpo = defined?(Reward) ? Reward.export_dpo : nil
  evalset = build_eval_set

  state = load_models
  version = state[:current].to_i + 1
  base = opts[:base_model] || (PWN::Env.dig(:ai, :ollama, :model) if defined?(PWN::Env)) || 'llama3'
  trainer = detect_trainer(preference: opts[:trainer])

  result = {
    version: version, base: base, trainer: trainer,
    sft: sft, dpo: dpo, eval_prompts: evalset.length, dry_run: dry_run
  }

  if dry_run || trainer.nil?
    result[:advice] = trainer.nil? ? "No trainer found. Install unsloth or axolotl, then re-run with dry_run:false. Datasets ready at #{sft&.[](:path)} + #{dpo&.[](:path)}." : 'dry_run — datasets + eval set exported; pass dry_run:false to train.'
    result[:manual_cli] = manual_train_cli(base: base, sft: sft, dpo: dpo, version: version)
    log(event: :train_and_gate, data: result)
    return result
  end

  adapter = run_trainer(trainer: trainer, base: base, sft: sft, dpo: dpo, version: version)
  return result.merge(error: 'trainer produced no adapter') unless adapter

  candidate = ollama_create(base: base, adapter: adapter, version: version)
  baseline  = state[:tag] || base
  gate = ab_gate(baseline: baseline, candidate: candidate, evalset: evalset)
  promoted = gate[:candidate_resolved] > gate[:baseline_resolved]
  if promoted
    state[:previous] = state[:tag]
    state[:tag] = candidate
    state[:current] = version
    state[:promoted_at] = Time.now.utc.iso8601
    state[:gate] = gate
    save_models(state: state)
  end
  result.merge(adapter: adapter, candidate: candidate, gate: gate, promoted: promoted)
rescue StandardError => e
  { error: "#{e.class}: #{e.message}" }
end