Module: PWN::AI::Agent::Reward

Defined in:

lib/pwn/ai/agent/reward.rb

Overview

PWN::AI::Agent::Reward is the OUTCOME reward model for the pwn-ai reinforcement-learning loop. It replaces the regex-proxy reward that previously drove Learning.infer_success / Loop.record_metrics with four calibrated signals:

R1  .judge      

Reward also owns the PREFERENCE-PAIR ledger (~/.pwn/preferences.jsonl) that turns pwn's naturally-generated (rejected, chosen) pairs — from user corrections, mistakes_resolve, and Curriculum.counterfactual A/B branches — into a DPO export (W1). This is the ONLY path from in-context learning to weight-level policy improvement.

E3 .verify_as_reward — grounds any final containing a checkable claim (CVE / version / cited URL) via Extrospection.verify and maps the browser verdict onto the reward scalar. Hallucination becomes a measurable −reward, not just a warning.

.judge prefers a cheap LLM ORM (direct engine .chat, short timeout, no Reflect / module_reflection gate). Reflect.on is used only when the operator enabled module_reflection (teacher engine). Heuristic token-overlap is LAST RESORT so proxy_distrust haircuts blend toward a real outcome signal, not bag-of-words overlap.

Constant Summary

PREFERENCES_FILE =

File.join(Dir.home, '.pwn', 'preferences.jsonl')

SENTINEL_FILE =

File.join(Dir.home, '.pwn', 'reward_sentinel.json')

DPO_DIR =

File.join(Dir.home, '.pwn', 'finetune')

SENTINEL_GAP =

0.15

SENTINEL_WINDOW =

40

VERDICTS =

{
  solved: 1.0, confirmed: 1.0, partial: 0.5,
  unknown: 0.5, wrong: 0.0, refused: 0.0, refuted: 0.0
}.freeze

BENIGN_EXIT =

Commands whose non-zero exit is INFORMATIONAL, not a failure. The regex-proxy treating these as failures was the single largest source of noise in Mistakes/Metrics (grep exit 1 = "no match").

{
  /\b(?:e|f|z|rip|p)?grep\b/ => [1],
  /\bdiff\b/ => [1],
  /\bcmp\b/ => [1],
  /\btest\b|\[\s/ => [1],
  /\bls\b/ => [1, 2],
  /\bfind\b/ => [1],
  /\bwhich\b|\bcommand -v\b/ => [1],
  /\bpidof\b|\bpgrep\b|\bpkill\b/ => [1],
  /\bxargs\b/ => [123],
  /\btimeout\b/ => [124],
  /\bcurl\b/ => [22],
  /\brubocop\b/ => [1]
}.freeze

JUDGE_SYSTEM =

"You are the pwn-ai Outcome Reward Model. Given a USER REQUEST, the\nagent's FINAL ANSWER, and a compressed TOOL TRACE, emit ONE line of\nstrict JSON:\n  {\"score\": <0.0-1.0>, \"verdict\": \"solved|partial|wrong|refused\",\n   \"rationale\": \"<\u2264140 chars>\", \"key_step\": <int|-1>}\nGrade the HUMAN RESULT against the USER REQUEST only \u2014 never a TUI\nplan, stub outline, or competing compass:\n  1.0 = final is usable and complete (every asked point answered with\n        evidence from the trace or a checkable claim).\n  0.7 = mostly complete, one missing detail, still usable.\n  0.5 = correct direction but incomplete / truncated.\n  0.2 = tools ran but the final does not answer the ask.\n  0.0 = hallucinated, off-goal, empty, polite non-answer, or refused.\nIgnore {\"success\":true} as evidence of done. Prefer last tool steps.\nkey_step is the 1-indexed shown-trace line most responsible, or -1.\nOutput JSON ONLY. No markdown fences.\n"

PRM_SYSTEM =

"You are the pwn-ai Process Reward Model. For EACH numbered tool\nstep, output one integer per line: 1 (advanced toward the goal),\n0 (neutral / exploratory), -1 (regressed / wasted). Output ONLY\nthe integers, one per line, same count as steps. No prose.\n"

TRAJECTORY_SHAPES =

Trajectory-shaped chosen sides that may land DPO without prose flood.

%w[winning_trace revised_answer real_dispatch].freeze

WRITE_SOURCE_CAP =

P9 — write-time source quota (not only export). Prefer diverse online generators over resolve-prose flood. Window is last WRITE_SOURCE_WINDOW pairs; a source already above WRITE_SOURCE_CAP is refused unless force: true (user_correction always forces).

0.40

WRITE_SOURCE_WINDOW =

100

TARGET_SOURCE_MIX =

P0 — target online generator mix for W1. Gates alone cannot fill an empty promote; the controller must prefer underfilled sources (counterfactual / critic / curriculum / user_correction) when resolve already dominates. Shares are soft targets, not hard caps (hard cap remains WRITE_SOURCE_CAP). Trajectory-only still applies.

{
  'mistakes_resolve' => 0.30,
  'curriculum' => 0.25,
  'counterfactual' => 0.20,
  'critic' => 0.15,
  'user_correction' => 0.10
}.freeze

DPO_SOURCE_CAP =

Max share any single preference source may occupy in a DPO export. Without this cap, mistakes_resolve monoculture (often >80%) teaches the LoRA "emit fix prose" instead of trajectory preference (P5 enforce).

0.40

CHEAP_ORM_TIMEOUT =

Cheap LLM ORM path. Reflect.on is gated by module_reflection (PII / teacher engine). Reward still needs a judge when that is off, so we call the active engine .chat directly with a short timeout. Fail fast to heuristic_judge rather than a 900s Reflect hang.

12

CHEAP_ORM_TEMP =

0.1

CHEAP_ORM_TRACE_N =

12

ORM_SAMPLE_WEIGHT =

1.0

HEURISTIC_SAMPLE_WEIGHT =

0.45

ERROR_SAMPLE_WEIGHT =

0.15

ENGINE_CHAT_MODS =

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

Class Method Summary

• .authors ⇒ Object

  Author(s)

  0day Inc.

• .clear_proxy_distrust ⇒ Object
• .export_dpo(opts = {}) ⇒ Object
• .generator_mix(opts = {}) ⇒ Object

  P0 — online generator mix report + urgency flags.

• .help ⇒ Object

  Display Usage for this Module.

• .infer_shape(opts = {}) ⇒ Object

  P0 ops — infer trajectory shape for legacy ledger rows that predate P21/P25 shape tags.

• .judge(opts = {}) ⇒ Object

  Supported Method Parameters

  v = PWN::AI::Agent::Reward.judge( request: 'required - original user request', final: 'required - assistant final answer', session_id: 'optional - PWN::Sessions id (adds tool trace)', trace: 'optional - Array of tool-result strings (overrides session_id)', commit: 'optional - write score into learning.jsonl / sentinel (default true)' ).

• .judge_sample_weight(opts = {}) ⇒ Object

  Weight a judge sample for sentinel / Learning haircuts.

• .plan_coverage(opts = {}) ⇒ Object

  ---------------------------------------------------------------- Plan-quality soft signal (W3 feature / Learning tag) ---------------------------------------------------------------- Cheap heuristic: did the final (+ optional tool trace) cover the tangible plan tasks? Not full DPO — trajectory-shaped pairs come later.

• .preference_balance(opts = {}) ⇒ Object

  P15/P5 — geometry-aware source mix.

• .preferences(opts = {}) ⇒ Object

  Supported Method Parameters

  rows = PWN::AI::Agent::Reward.preferences(limit: 500, source: nil).

• .prm(opts = {}) ⇒ Object

  Supported Method Parameters

  steps = PWN::AI::Agent::Reward.prm( request: 'required - user goal', session_id: 'optional - session to score in place', trace: 'optional - Array of args:, result: or Strings' ).

• .promote_to_success?(opts = {}) ⇒ Boolean
• .proxy_distrust ⇒ Object

  P4 — scalar 0.0..1.0 haircut applied to Metrics success / Registry β when the proxy is lying.

• .record_preference(opts = {}) ⇒ Object
• .record_verification(opts = {}) ⇒ Object

  Trusted host-verifier API, deliberately NOT a model-facing tool.

• .recoverable_shape(opts = {}) ⇒ Object

  2.2 — coarse recoverable shape beside the fingerprint.

• .reset ⇒ Object
• .reset_sentinel ⇒ Object

  One-shot: wipe sentinel window + distrust after deploying the ring-buffer arithmetic (or any time the live file is known-corrupt).

• .resolve_outcome(opts = {}) ⇒ Object

  The sole outcome decision.

• .run_verification(opts = {}) ⇒ Object

  Trusted host-verifier API, deliberately NOT a model-facing tool.

• .scrub_preferences(opts = {}) ⇒ Object

  P15 — one-shot ledger hygiene.

• .semantic_ok(opts = {}) ⇒ Object

  Supported Method Parameters

  h = PWN::AI::Agent::Reward.semantic_ok( name: 'required - tool name', raw: 'required - JSON string returned by Dispatch.call', args: 'optional - the tool call arguments (used for BENIGN_EXIT)' ).

• .sentinel ⇒ Object

  Supported Method Parameters

  r = PWN::AI::Agent::Reward.sentinel.

• .set_proxy_distrust(opts = {}) ⇒ Object
• .usable_preference?(opts = {}) ⇒ Boolean

  P15 — keep only usable preference pairs for balance/export/promote.

• .verify_as_reward(opts = {}) ⇒ Object

  Supported Method Parameters

  g = PWN::AI::Agent::Reward.verify_as_reward(final: text).

• .warm_sentinel(opts = {}) ⇒ Object

  P10 — backfill the R3 ring from Learning outcomes so offline/local hosts reach SENTINEL_WINDOW without waiting for live remote introspect.

• .write_source_quota(opts = {}) ⇒ Object

  Share of source among the newest WRITE_SOURCE_WINDOW prefs.

Class Method Details

.authors ⇒ Object

Author(s)

0day Inc. support@0dayinc.com

# File 'lib/pwn/ai/agent/reward.rb', line 1955

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

.clear_proxy_distrust ⇒ Object

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

public_class_method def self.clear_proxy_distrust
  s = load_sentinel
  return if s[:proxy_distrust].to_f <= 0.0

  s[:proxy_distrust] = 0.0
  s[:distrust_cleared_at] = Time.now.utc.iso8601
  atomic_write(path: SENTINEL_FILE, body: JSON.generate(s))
rescue StandardError
  nil
end

.export_dpo(opts = {}) ⇒ Object

# File 'lib/pwn/ai/agent/reward.rb', line 1172

public_class_method def self.export_dpo(opts = {})
  fmt = (opts[:format] || :dpo).to_sym
  FileUtils.mkdir_p(DPO_DIR)
  out = opts[:out] || File.join(DPO_DIR, "pwn-dpo-#{Time.now.utc.strftime('%Y%m%d')}.jsonl")
  rows = preferences(limit: 100_000)
  # P15 — drop weak geometry before source-cap so resolve prose cannot
  # dominate the kept set after balance. opt-out with scrub: false.
  scrub = opts.key?(:scrub) ? opts[:scrub] : true
  geometry_dropped = 0
  if scrub
    usable = rows.select { |r| usable_preference?(row: r) }
    geometry_dropped = rows.length - usable.length
    rows = usable
  end
  # P5 — downsample so no single source exceeds DPO_SOURCE_CAP of the export.
  # opt-out with balance: false (raw dump for diagnostics).
  balance = opts.key?(:balance) ? opts[:balance] : true
  selected = balance ? balance_preference_rows(rows: rows, cap: (opts[:source_cap] || DPO_SOURCE_CAP).to_f) : rows
  dropped = rows.length - selected.length
  File.open(out, 'w') do |f|
    selected.each do |r|
      line = case fmt
             when :kto
               [{ prompt: r[:prompt], completion: r[:chosen], label: true },
                { prompt: r[:prompt], completion: r[:rejected], label: false }]
             else
               # Keep source for auditability / preference_balance post-export.
               { prompt: r[:prompt], chosen: r[:chosen], rejected: r[:rejected], source: r[:source] }
             end
      (line.is_a?(Array) ? line : [line]).each { |l| f.puts(JSON.generate(l)) }
    end
  end
  by_src = selected.group_by { |r| r[:source].to_s }.transform_values(&:length)
  {
    path: out, format: fmt, pairs: selected.length, bytes: File.size(out),
    balanced: balance, dropped: dropped, geometry_dropped: geometry_dropped,
    scrubbed: scrub, by_source: by_src,
    source_cap: balance ? (opts[:source_cap] || DPO_SOURCE_CAP).to_f : nil,
    preference_balance: begin
      preference_balance(limit: 10_000, scrub: true)
    rescue StandardError
      nil
    end
  }
end

.generator_mix(opts = {}) ⇒ Object

P0 — online generator mix report + urgency flags. Controllers (auto_introspect, practice, counterfactual gate) consult this so underfilled sources get scheduling priority while over-cap resolve stops flooding. Returns trajectory_fraction, urgent:[], suppress:[], healthy:.

# File 'lib/pwn/ai/agent/reward.rb', line 923

public_class_method def self.generator_mix(opts = {})
  limit = opts[:limit] || WRITE_SOURCE_WINDOW
  rows = preferences(limit: limit)
  usable = rows.select { |r| usable_preference?(row: r) }
  by = Hash.new(0)
  usable.each { |r| by[r[:source].to_s] += 1 }
  n = usable.length
  shares = {}
  TARGET_SOURCE_MIX.each_key { |k| shares[k] = n.zero? ? 0.0 : (by[k].to_f / n).round(3) }
  by.each_key { |k| shares[k] ||= (by[k].to_f / n).round(3) }

  traj_n = usable.count { |r| TRAJECTORY_SHAPES.include?(r[:shape].to_s) }
  traj_f = n.zero? ? 0.0 : (traj_n.to_f / n).round(3)

  urgent = []
  suppress = []
  TARGET_SOURCE_MIX.each do |src, target|
    sh = shares[src].to_f
    urgent << src if sh < (target * 0.5) && n >= 5
    suppress << src if sh > WRITE_SOURCE_CAP && n >= 10
  end
  suppress << 'mistakes_resolve' if shares['mistakes_resolve'].to_f > WRITE_SOURCE_CAP && n >= 10 && !suppress.include?('mistakes_resolve')

  healthy = urgent.empty? && suppress.empty? && traj_f >= 0.5 && n >= 10
  {
    n: n,
    raw_n: rows.length,
    by_source: by,
    shares: shares,
    targets: TARGET_SOURCE_MIX,
    trajectory_fraction: traj_f,
    urgent: urgent.uniq,
    suppress: suppress.uniq,
    healthy: healthy,
    recommendation: if healthy
                      'mix_ok'
                    elsif n < 10
                      'need_more_pairs'
                    elsif traj_f < 0.5
                      'need_trajectory_shape'
                    elsif urgent.any?
                      "boost:#{urgent.join(',')}"
                    else
                      "suppress:#{suppress.join(',')}"
                    end
  }
rescue StandardError => e
  {
    n: 0, healthy: false, error: "#{e.class}: #{e.message}",
    urgent: %w[curriculum counterfactual critic user_correction],
    suppress: []
  }
end

.help ⇒ Object

Display Usage for this Module

# File 'lib/pwn/ai/agent/reward.rb', line 1961

public_class_method def self.help
  puts "USAGE:
    # R1 — LLM Outcome Reward Model
    #{self}.judge(
      request: 'required - original user request',
      final: 'required - assistant final answer',
      session_id: 'optional - PWN::Sessions id (adds tool trace)',
      trace: 'optional - Array of tool-result strings (overrides session_id)',
      commit: 'optional - write score into learning.jsonl / sentinel (default true)',
      critic_pass: 'optional - critic pass value consumed by #judge',
      predicted: 'optional - predicted value consumed by #judge',
      proxy_ok: 'optional - proxy ok value consumed by #judge',
      persist_components: 'optional - persist the resolved outcome and score components',
      verification_contract: 'optional - explicit host-owned Verification.run contract; missing coverage stays unknown',
      verification: 'optional - Hash with verdict pass and evidence of at least 40 characters'
    )

    # Execute host-owned acceptance checks and record request-bound coverage.
    #{self}.run_verification(
      request: 'required - exact current session request',
      session_id: 'required - current session',
      contract: 'required - Verification.run options excluding request',
      commit: 'optional - persist report (default true)'
    )

    # Trusted host verifier only: actually check ALL original-request criteria.
    # PASS prose, exit zero, and single confirmed claims are not substitutes.
    #{self}.record_verification(
      request: 'required - exact current session user request',
      session_id: 'required - session holding the request and verification',
      checks: 'required - complete Array of {criterion:, passed: Boolean, evidence:}'
    )

    # Shared verdict and training eligibility; unknown training_score is nil.
    #{self}.resolve_outcome(
      outcome: 'required - outcome Hash from the judge or trusted evaluator',
      critic_pass: 'optional - false records a critic disagreement'
    )

    # Run promote to success and return its result
    #{self}.promote_to_success?(
      orm: 'optional - orm value consumed by #promote_to_success?',
      verify: 'optional - verify value consumed by #promote_to_success?',
      critic: 'optional - critic value consumed by #promote_to_success?'
    )

    # Run prm and return its result
    #{self}.prm(
      request: 'required - user goal',
      session_id: 'optional - session to score in place',
      trace: 'optional - Array of {name:, args:, result:} or Strings'
    )

    # Plan-quality soft signal (W3 feature / Learning tag)
    #{self}.plan_coverage(
      plan: 'required - Array of task strings or outline text',
      final: 'required - assistant final answer',
      request: 'optional - original user request',
      trace: 'optional - Array of tool-result strings',
      session_id: 'optional - load trace from session when trace empty'
    )

    # R3 — Reward-hacking sentinel
    #{self}.sentinel

    # P4 — scalar 0.0..1.0 haircut applied to Metrics success / Registry β when
    #{self}.proxy_distrust

    # Run set proxy distrust and return its result
    #{self}.set_proxy_distrust(
      gap: 'optional - gap value consumed by #set_proxy_distrust',
      proxy: 'optional - scheme://proxy_host:port, or tor',
      judge: 'optional - judge value consumed by #set_proxy_distrust'
    )

    # Run clear proxy distrust and return its result
    #{self}.clear_proxy_distrust

    # One-shot: wipe sentinel window + distrust after deploying the
    #{self}.reset_sentinel

    # P10 — backfill the R3 ring from Learning outcomes so offline/local
    #{self}.warm_sentinel(
      limit: 'optional - limit value consumed by #warm_sentinel'
    )

    # R4 — Structured tool-result classifier
    #{self}.semantic_ok(
      name: 'required - tool name',
      raw: 'required - JSON string returned by Dispatch.call',
      args: 'optional - the tool call arguments (used for BENIGN_EXIT)'
    )

    # 2.2 — coarse recoverable shape beside the fingerprint. Paths are
    #{self}.recoverable_shape(
      err: 'optional - err value consumed by #recoverable_shape',
      stderr: 'optional - stderr value consumed by #recoverable_shape',
      exit_code: 'optional - exit code value consumed by #recoverable_shape'
    )

    # E3 — verify-as-reward (ground truth without a human)
    #{self}.verify_as_reward(
      final: 'optional - final value consumed by #verify_as_reward'
    )

    # Run record preference and return its result
    #{self}.record_preference(
      prompt: 'required - prompt value consumed by #record_preference',
      rejected: 'required - rejected value consumed by #record_preference',
      chosen: 'required - chosen value consumed by #record_preference',
      force: 'optional - force value consumed by #record_preference',
      source: 'optional - source value consumed by #record_preference (defaults to :unknown))',
      shape: 'optional - shape value consumed by #record_preference',
      meta: 'optional - meta value consumed by #record_preference'
    )

    # Share of `source` among the newest WRITE_SOURCE_WINDOW prefs
    #{self}.write_source_quota(
      source: 'optional - source value consumed by #write_source_quota'
    )

    # P0 — online generator mix report + urgency flags. Controllers
    #{self}.generator_mix(
      limit: 'optional - limit value consumed by #generator_mix (defaults to WRITE_SOURCE_WINDOW)'
    )

    # P0 ops — infer trajectory shape for legacy ledger rows that predate
    #{self}.infer_shape(
      row: 'optional - row value consumed by #infer_shape'
    )

    # P15 — keep only usable preference pairs for balance/export/promote
    #{self}.usable_preference?(
      row: 'optional - row value consumed by #usable_preference?'
    )

    # P15 — one-shot ledger hygiene. Filters in place (rewrite jsonl) or
    #{self}.scrub_preferences(
      dry_run: 'optional - dry run value consumed by #scrub_preferences'
    )

    # P15/P5 — geometry-aware source mix. scrub:true uses usable_preference?
    #{self}.preference_balance(
      limit: 'optional - limit value consumed by #preference_balance (defaults to 10_000)',
      scrub: 'optional - scrub value consumed by #preference_balance'
    )

    # Run preferences and return its result
    #{self}.preferences(
      limit: 'optional - limit value consumed by #preferences (defaults to 500)',
      source: 'required - source value consumed by #preferences'
    )

    # Run export dpo and return its result
    #{self}.export_dpo(
      format: 'optional - format value consumed by #export_dpo (defaults to :dpo))',
      out: 'optional - out value consumed by #export_dpo',
      scrub: 'optional - scrub value consumed by #export_dpo',
      balance: 'optional - balance value consumed by #export_dpo',
      source_cap: 'optional - source cap value consumed by #export_dpo'
    )

    # Run reset and return its result
    #{self}.reset

    # Weight a judge sample for sentinel / Learning haircuts
    #{self}.judge_sample_weight(
      source: 'optional - source value consumed by #judge_sample_weight'
    )

    # Print the AUTHOR(S) string for this module.
    #{self}.authors
  "
  constants.sort
end

.infer_shape(opts = {}) ⇒ Object

P0 ops — infer trajectory shape for legacy ledger rows that predate P21/P25 shape tags. Used by scrub_preferences rewrite so generator_mix trajectory_fraction reflects content, not missing keys.

# File 'lib/pwn/ai/agent/reward.rb', line 980

public_class_method def self.infer_shape(opts = {})
  r = opts.is_a?(Hash) && opts.key?(:row) ? opts[:row] : opts
  r = r.transform_keys(&:to_sym) if r.respond_to?(:transform_keys)
  existing = r[:shape].to_s
  return existing if TRAJECTORY_SHAPES.include?(existing) || existing == 'fix_prose'

  chosen = r[:chosen].to_s
  source = r[:source].to_s
  # tool-call / trace markers → winning_trace
  if chosen.match?(/\b(shell|pwn_eval|memory_|sessions_|reward_|curriculum_|extro_|mistakes_)\b/i) &&
     (chosen.include?('→') || chosen.include?('tool_call') || chosen.include?('"name"') ||
      chosen.lines.count { |l| l.strip.start_with?('{') || l.include?('arguments') } >= 1)
    return 'winning_trace'
  end
  # long revised answer from critic / user / CF → revised_answer
  return 'revised_answer' if chosen.length >= 200 && %w[critic user_correction counterfactual curriculum].include?(source)

  # counterfactual real dispatch tag in meta
  meta = r[:meta].is_a?(Hash) ? r[:meta] : {}
  return 'real_dispatch' if meta[:mode].to_s == 'real_dispatch' || meta['mode'].to_s == 'real_dispatch'

  existing.empty? ? nil : existing
rescue StandardError
  nil
end

.judge(opts = {}) ⇒ Object

Supported Method Parameters

v = PWN::AI::Agent::Reward.judge( request: 'required - original user request', final: 'required - assistant final answer', session_id: 'optional - PWN::Sessions id (adds tool trace)', trace: 'optional - Array of tool-result strings (overrides session_id)', commit: 'optional - write score into learning.jsonl / sentinel (default true)' )

# File 'lib/pwn/ai/agent/reward.rb', line 121

public_class_method def self.judge(opts = {})
  request = opts[:request].to_s
  final   = opts[:final].to_s
  trace   = Array(opts[:trace])
  trace   = load_trace(session_id: opts[:session_id]) if trace.empty? && opts[:session_id]
  commit  = opts.key?(:commit) ? opts[:commit] : true

  v = llm_judge(request: request, final: final, trace: trace)
  v ||= heuristic_judge(request: request, final: final, trace: trace)
  # Cheap ORM is the intended source. Heuristic overlap is fallback
  # only — callers (sentinel / Learning.stats / Metrics.effective_rate)
  # weight :llm_orm samples above :heuristic so the haircut tracks
  # the outcome model, not token overlap.

  # P1 — local/heuristic calibration: thin judges must not be treated
  # as ground truth when proxy_distrust is already high. Two levers:
  #   (1) low :confidence so Metrics.effective_rate haircuts blend
  #       weight (distrust × confidence) instead of replacing proxy;
  #   (2) score-path caps only for decisive failure floors and for
  #       local no-trace highs (false "solved"). Do NOT pull known
  #       wrong (0.0 failure-language) toward 0.5, and do NOT deflate
  #       tool-backed heuristic scores that already cleared the bar —
  #       confidence handles that in the bandit blend.
  eng = (PWN::Env.dig(:ai, :active) if defined?(PWN::Env)).to_s.downcase
  local = eng == 'ollama' || eng.empty?
  if v[:source].to_s == 'heuristic' || v[:source].to_s.start_with?('heuristic')
    toolbacked = Array(trace).any?
    v[:heuristic_class] = toolbacked ? :toolbacked : :textual
    v[:source] = :heuristic
    v[:confidence] = if toolbacked
                       local ? 0.55 : 0.7
                     else
                       local ? 0.35 : 0.5
                     end
    raw = v[:score].to_f
    v[:score_raw] = raw
    if !toolbacked && local && raw > 0.15 && trace.empty? && raw >= 0.6 && final.length < 400
      v[:score] = [raw, 0.45].min
      v[:rationale] = "#{v[:rationale]} | P1:local_no_trace_cap"
    elsif !toolbacked && local && raw > 0.15 && trace.length < 2 && raw >= 0.85
      v[:score] = (0.5 + ((raw - 0.5) * 0.7)).round(3).clamp(0.0, 1.0)
    else
      v[:score] = raw
    end

    v[:verdict] = if v[:score] >= 0.6 then :solved
                  elsif v[:score] >= 0.3 then :partial
                  else :wrong
                  end
  else
    v[:confidence] ||= 0.85
  end

  ground = verify_as_reward(final: final)
  unless ground.nil?
    # A refuted claim is negative evidence. Confirming one claim
    # does not establish completion of the entire operator request.
    v[:score] = [v[:score], 0.2].min if ground[:verdict] == :refuted
    v[:grounded] = ground
    v[:confidence] = [v[:confidence].to_f, ground[:confidence].to_f].max if ground[:confidence]
  end

  v[:judge_score] = v[:score].to_f
  v[:score_quality] = v[:score].to_f
  v[:verification] = if opts[:verification].is_a?(Hash)
                       evidence_pass_verification(verification: opts[:verification])
                     elsif opts[:verification_contract]
                       run_verification(request: request, session_id: opts[:session_id], contract: opts[:verification_contract], commit: commit)
                     else
                       request_verification(request: request, session_id: opts[:session_id])
                     end
  v = resolve_outcome(outcome: v, critic_pass: opts[:critic_pass])
  v[:rationale] = "#{v[:rationale]} | verification_floor" if opts[:verification].is_a?(Hash) && v[:success] == true
  v[:verdict_class] = taxonomy_class(opts.merge(score: v[:score], verifier_verdict: v[:verifier_verdict], request: request, final: final))
  v[:remediation_hint] = taxonomy_hint(verdict_class: v[:verdict_class])
  v[:needs_spot_check] = v[:success] && v[:score].to_f >= 0.85 && (rand < 0.05)
  v[:engine] = eng
  v[:task_class] = request.match?(/analy[sz]e|summar|strength|weakness|fitness/i) ? 'analysis' : 'operational'
  v[:score_components] ||= {
    judge: v[:score].to_f,
    overlap: nil,
    checks: 0.0,
    weights: { overlap: 0.15 }
  }
  if commit && defined?(Learning) && opts[:persist_components]
    Learning.note_outcome(
      task: request[0, 80],
      success: v[:success],
      score: v[:score],
      outcome: v,
      details: v[:score_components].to_json,
      verifier_verdict: v[:verifier_verdict],
      verdict_class: v[:verdict_class]
    )
  end
  # W3 — write Brier on every judged turn so overconfidence can
  # throttle max_iters/critic even when plan_first never fired.
  if commit && !v[:training_score].nil?
    pred = opts[:predicted]
    pred = Thread.current[:pwn_plan_predicted] if pred.nil?
    pred = v[:confidence] if pred.nil?
    Curriculum.calibrate(predicted: pred, actual: v[:score], engine: eng) if defined?(Curriculum) && Curriculum.respond_to?(:calibrate)
  end
  # P1 — sentinel stores confidence so distrust math can haircut
  # heuristic-heavy windows differently from LLM ORM windows.
  record_sentinel(v.slice(:training_score, :decision_version, :verdict, :confidence, :source).merge(proxy: opts[:proxy_ok], judge: v[:training_score])) if commit && !v[:training_score].nil?
  v
rescue StandardError => e
  resolve_outcome(outcome: { score: nil, rationale: "judge error: #{e.class}", error: e.message, confidence: 0.0, source: :error })
end

.judge_sample_weight(opts = {}) ⇒ Object

Weight a judge sample for sentinel / Learning haircuts. LLM ORM counts as a full outcome; heuristic overlap is a cheap prior so it cannot dominate proxy_distrust when real ORM exists.

# File 'lib/pwn/ai/agent/reward.rb', line 1250

public_class_method def self.judge_sample_weight(opts = {})
  case opts[:source].to_s
  when 'llm_orm' then ORM_SAMPLE_WEIGHT
  when 'error' then ERROR_SAMPLE_WEIGHT
  else HEURISTIC_SAMPLE_WEIGHT
  end
end

.plan_coverage(opts = {}) ⇒ Object

---

Plan-quality soft signal (W3 feature / Learning tag)

Cheap heuristic: did the final (+ optional tool trace) cover the tangible plan tasks? Not full DPO — trajectory-shaped pairs come later. Score is a soft feature for calibration / tagging only.

Supported Method Parameters

r = PWN::AI::Agent::Reward.plan_coverage( plan: 'required - Array of task strings or outline text', final: 'required - assistant final answer', request: 'optional - original user request', trace: 'optional - Array of tool-result strings', session_id: 'optional - load trace from session when trace empty' ) => { score: 0.0..1.0, covered: N, total: M, missing: [...], tag: 'plan_cover_high|mid|low' }

# File 'lib/pwn/ai/agent/reward.rb', line 435

public_class_method def self.plan_coverage(opts = {})
  plan = opts[:plan]
  tasks =
    case plan
    when Array then plan.map(&:to_s)
    when String
      if defined?(TaskSummarizer) && TaskSummarizer.respond_to?(:parse_outline_tasks)
        TaskSummarizer.parse_outline_tasks(outline: plan)
      else
        plan.to_s.split(/\n+/).map { |l| l.sub(/\A(?:\d+[.):]|[-*•])\s+/, '').strip }
      end
    else
      Array(plan).map(&:to_s)
    end
  tasks = tasks.map { |t| t.to_s.gsub(/\s+/, ' ').strip }.reject(&:empty?)
  return { score: 0.0, covered: 0, total: 0, missing: [], tag: 'plan_cover_none' } if tasks.empty?

  final = opts[:final].to_s
  request = opts[:request].to_s
  trace = Array(opts[:trace])
  trace = load_trace(session_id: opts[:session_id]) if trace.empty? && opts[:session_id]
  blob = "#{final}\n#{request}\n#{trace.join("\n")}".downcase

  covered = []
  missing = []
  tasks.each do |task|
    stems = task.downcase.scan(/[a-z0-9]{4,}/).uniq
    # Drop ultra-generic plan fillers that would false-positive everything.
    stems.reject! { |s| %w[result results report verify complete completion present carry work task step this that with from into].include?(s) }
    if stems.empty?
      covered << task
      next
    end
    # A task is covered when >= half of its distinctive stems appear
    # in final+trace (soft — not DPO-grade evidence).
    hits = stems.count { |s| blob.include?(s) }
    need = [1, (stems.length / 2.0).ceil].max
    if hits >= need
      covered << task
    else
      missing << task
    end
  end

  total = tasks.length
  score = (covered.length.to_f / total).round(3).clamp(0.0, 1.0)
  tag =
    if score >= 0.75 then 'plan_cover_high'
    elsif score >= 0.4 then 'plan_cover_mid'
    else 'plan_cover_low'
    end
  {
    score: score,
    covered: covered.length,
    total: total,
    missing: missing.first(6),
    tag: tag
  }
rescue StandardError
  { score: 0.0, covered: 0, total: 0, missing: [], tag: 'plan_cover_error' }
end

.preference_balance(opts = {}) ⇒ Object

P15/P5 — geometry-aware source mix. scrub:true uses usable_preference? so operators see the post-hygiene diet (what export_dpo will train on).

# File 'lib/pwn/ai/agent/reward.rb', line 1090

public_class_method def self.preference_balance(opts = {})
  limit = opts[:limit] || 10_000
  scrub = opts.key?(:scrub) ? opts[:scrub] : false
  rows = preferences(limit: limit)
  before = rows.length
  rows = rows.select { |r| usable_preference?(row: r) } if scrub
  by = Hash.new(0)
  by_shape = Hash.new(0)
  rows.each do |r|
    by[r[:source].to_s] += 1
    sh = r[:shape].to_s
    sh = 'unspecified' if sh.empty?
    by_shape[sh] += 1
  end
  total = rows.length
  frac = by.transform_values { |n| total.zero? ? 0.0 : (n.to_f / total).round(3) }
  shape_frac = by_shape.transform_values { |n| total.zero? ? 0.0 : (n.to_f / total).round(3) }
  traj_n = rows.count { |r| TRAJECTORY_SHAPES.include?(r[:shape].to_s) }
  traj_frac = total.zero? ? 0.0 : (traj_n.to_f / total).round(3)
  monoculture = total.positive? && (by.values.max.to_f / total) > 0.7
  mix = begin
    generator_mix(limit: limit)
  rescue StandardError
    nil
  end
  {
    total: before,
    kept: total,
    scrubbed: scrub,
    dropped: before - total,
    by_source: by,
    fractions: frac,
    by_shape: by_shape,
    by_shape_fraction: shape_frac,
    trajectory_fraction: traj_frac,
    monoculture: monoculture,
    generator_mix: mix,
    advice: if total < 12
              'W1 thin: need more trajectory-shaped pairs before LoRA promote.'
            elsif monoculture
              'W1 monoculture: run Reward.scrub_preferences; enable :counterfactual/:critic; stop resolve-prose flood.'
            elsif traj_frac < 0.30
              'W1 geometry weak: <30% trajectory-shaped chosen sides — DPO would teach commentary.'
            elsif mix && !mix[:healthy]
              "W1 generator mix: #{mix[:recommendation]}"
            else
              'W1 source mix OK for gated export'
            end
  }
rescue StandardError => e
  { error: "#{e.class}: #{e.message}" }
end

.preferences(opts = {}) ⇒ Object

Supported Method Parameters

rows = PWN::AI::Agent::Reward.preferences(limit: 500, source: nil)

# File 'lib/pwn/ai/agent/reward.rb', line 1146

public_class_method def self.preferences(opts = {})
  limit  = opts[:limit] || 500
  source = opts[:source].to_s
  return [] unless File.exist?(PREFERENCES_FILE)

  rows = File.readlines(PREFERENCES_FILE).map do |l|
    JSON.parse(l, symbolize_names: true)
  rescue StandardError
    nil
  end
  rows.compact!
  rows.select! { |r| r[:source] == source } unless source.empty?
  rows.reverse.first(limit)
end

.prm(opts = {}) ⇒ Object

Supported Method Parameters

steps = PWN::AI::Agent::Reward.prm( request: 'required - user goal', session_id: 'optional - session to score in place', trace: 'optional - Array of args:, result: or Strings' )

Returns [step:, reward: -1|0|1, ...] and, when session_id is given, rewrites each tool line in the transcript with a [step_reward=N] prefix so exemplars_for / distill_skill can keep only reward>0 steps (C4 minimal sufficient trace).

# File 'lib/pwn/ai/agent/reward.rb', line 401

public_class_method def self.prm(opts = {})
  request = opts[:request].to_s
  trace   = Array(opts[:trace])
  sid     = opts[:session_id]
  trace   = load_trace(session_id: sid) if trace.empty? && sid

  rewards = llm_prm(request: request, trace: trace)
  rewards ||= heuristic_prm(trace: trace)

  out = trace.each_with_index.map do |s, i|
    { idx: i + 1, step: s.to_s[0, 200], reward: rewards[i] || 0 }
  end
  annotate_session(session_id: sid, rewards: rewards) if sid
  out
rescue StandardError
  []
end

.promote_to_success?(opts = {}) ⇒ Boolean

Returns:

• (Boolean)

# File 'lib/pwn/ai/agent/reward.rb', line 378

public_class_method def self.promote_to_success?(opts = {})
  flags = []
  flags << (opts[:orm] ? true : false) unless opts[:orm].nil?
  flags << (opts[:verify] ? true : false) unless opts[:verify].nil?
  flags << (opts[:critic] ? true : false) unless opts[:critic].nil?
  return false if flags.empty?
  return flags.first if flags.length == 1

  flags.count(true) >= 2
end

.proxy_distrust ⇒ Object

P4 — scalar 0.0..1.0 haircut applied to Metrics success / Registry β when the proxy is lying. 0.0 = trust proxy fully; 1.0 = ignore proxy rates.

# File 'lib/pwn/ai/agent/reward.rb', line 568

public_class_method def self.proxy_distrust
  s = load_sentinel
  d = s[:proxy_distrust].to_f
  # auto-expire after 7d without refresh so a one-off gap doesn't stick
  if s[:distrust_at]
    age = Time.now.utc - Time.parse(s[:distrust_at].to_s)
    return 0.0 if age > 7 * 86_400
  end
  d = d.clamp(0.0, 1.0)
  # Recalibrated cap: leftover 1.0 from the old mapping must not fully
  # haircut raw success unless the live gap is still extreme.
  meta = s[:distrust_meta] || {}
  gap = (meta[:gap] || meta['gap']).to_f
  [d, 0.85].min
rescue StandardError
  0.0
end

.record_preference(opts = {}) ⇒ Object

# File 'lib/pwn/ai/agent/reward.rb', line 837

public_class_method def self.record_preference(opts = {})
  prompt   = opts[:prompt].to_s
  rejected = opts[:rejected].to_s
  chosen   = opts[:chosen].to_s
  return nil if prompt.strip.empty? || chosen.strip.empty? || rejected.strip.empty?
  return nil if chosen.strip == rejected.strip

  # Reject weak pair geometry: CORRECTION: flaw-prose is not a trajectory.
  return { skipped: :weak_pair_geometry, reason: 'chosen looks like flaw prose, not a revised answer/trace' } if chosen.match?(/\A\s*CORRECTION:\s*/i) && chosen.length < 400 && !opts[:force]

  source = (opts[:source] || :unknown).to_s
  shape  = opts[:shape].to_s
  # P25 — require trajectory shape at write time unless force / user_correction.
  # Stops resolve-prose flood from ever landing in the ledger; export scrub
  # is defense-in-depth, not the primary gate.
  traj = TRAJECTORY_SHAPES.include?(shape)
  # P25 — non-trajectory prose never lands (export scrub is defense-in-depth).
  # user_correction and explicit force: still allowed for human / migration paths.
  unless traj || opts[:force] || source == 'user_correction'
    return {
      skipped: :non_trajectory_shape,
      reason: "shape=#{shape.inspect} not in #{TRAJECTORY_SHAPES.join(',')}; pass force:true or a trajectory shape",
      source: source
    }
  end
  # P9 — write-time source quota still applies to trajectory pairs.
  # P25 made every auto-written row trajectory-shaped; if traj also
  # bypassed the quota, resolve monoculture would return via winning_trace
  # flood. Only user_correction and explicit force:true skip the cap.
  bypass_quota = opts[:force] || source == 'user_correction'
  unless bypass_quota
    quota = write_source_quota(source: source)
    return quota.merge(skipped: :source_quota) if quota[:over_cap]

    # P0 — also refuse sources the live mix already asked to suppress
    # (critic 40% vs 15% target) so write-time, not only export, rebalances.
    mix = generator_mix
    return quota.merge(skipped: :source_quota, over_cap: true, reason: "mix_suppress:#{source}") if Array(mix[:suppress]).include?(source) && mix[:n].to_i >= 10
  end

  entry = {
    id: Digest::SHA256.hexdigest("#{prompt}|#{rejected}|#{chosen}")[0, 12],
    prompt: prompt[0, 4_000],
    rejected: rejected[0, 4_000],
    chosen: chosen[0, 4_000],
    source: source,
    engine: (PWN::Env.dig(:ai, :active) if defined?(PWN::Env)).to_s,
    timestamp: Time.now.utc.iso8601
  }
  entry[:meta] = opts[:meta] if opts[:meta].is_a?(Hash)
  entry[:shape] = opts[:shape].to_s if opts[:shape]
  FileUtils.mkdir_p(File.dirname(PREFERENCES_FILE))
  File.open(PREFERENCES_FILE, 'a') { |f| f.puts(JSON.generate(entry)) }
  entry
end

.record_verification(opts = {}) ⇒ Object

Trusted host-verifier API, deliberately NOT a model-facing tool. Caller must actually check every original-request criterion. Never construct these records by parsing an assistant/tool claim of PASS.

Raises:

• (ArgumentError)

# File 'lib/pwn/ai/agent/reward.rb', line 325

public_class_method def self.record_verification(opts = {})
  request = opts[:request].to_s
  sid = opts[:session_id].to_s
  checks = opts[:checks]
  raise ArgumentError, 'nonempty complete request checks required' unless valid_verification_checks?(checks: checks)

  rows = PWN::Sessions.load(session_id: sid)
  user = rows.reverse.find { |row| row[:role].to_s == 'user' }
  raise ArgumentError, 'verification must match the current session request' if request.empty? || !user || user[:content].to_s != request

  record = { request_digest: Digest::SHA256.hexdigest(request), session_id: sid, requirements: [request], checks: checks }
  PWN::Sessions.append(session_id: sid, role: 'verification', content: JSON.generate(record))
  record
end

.recoverable_shape(opts = {}) ⇒ Object

2.2 — coarse recoverable shape beside the fingerprint. Paths are normalised away for counting; shape stays for repair routing (enoent → install/check path; exit127 → missing binary; …).

# File 'lib/pwn/ai/agent/reward.rb', line 744

public_class_method def self.recoverable_shape(opts = {})
  err = "#{opts[:err]} #{opts[:stderr]}".downcase
  ec  = opts[:exit_code]
  return :exit127 if ec == 127 || err.include?('command not found')
  return :exit126 if ec == 126
  return :enoent if err.match?(/no such file|enoent|cannot access|not a directory/)
  return :eacces if err.match?(/permission denied|eacces|operation not permitted/)
  return :auth_required if err.match?(/auth|unauthorized|401|403|forbidden|login required|api.?key/)
  return :timeout if ec == 124 || err.include?('timed out') || err.include?('timeout')
  return :network if err.match?(/connection refused|name or service not known|could not resolve|network is unreachable/)
  return :syntax if err.match?(/syntax error|parse error|unexpected token|json::parser/)
  return :nonzero_exit if ec && ec != 0
  return :handler_error if err.strip.length.positive?

  :unknown
end

.reset ⇒ Object

# File 'lib/pwn/ai/agent/reward.rb', line 1218

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

.reset_sentinel ⇒ Object

One-shot: wipe sentinel window + distrust after deploying the ring-buffer arithmetic (or any time the live file is known-corrupt). Does NOT touch preferences / DPO exports (unlike .reset).

# File 'lib/pwn/ai/agent/reward.rb', line 623

public_class_method def self.reset_sentinel
  FileUtils.rm_f(SENTINEL_FILE)
  { cleared: true, path: SENTINEL_FILE }
end

.resolve_outcome(opts = {}) ⇒ Object

The sole outcome decision. Scores are diagnostic; training_score is absent when the evaluator cannot supply a reliable label.

# File 'lib/pwn/ai/agent/reward.rb', line 249

public_class_method def self.resolve_outcome(opts = {})
  v = (opts[:outcome] || {}).dup
  source = (v[:source] || v[:judge_source]).to_s
  score = v[:score]
  score = score.to_f.clamp(0.0, 1.0) unless score.nil?
  v[:quality_score] = v.fetch(:judge_score, score) unless v.key?(:quality_score)
  verification = v[:verification]
  checked = verification.is_a?(Hash) && valid_verification_checks?(checks: verification[:checks])
  requirements = verification.is_a?(Hash) ? Array(verification[:requirements]) : []
  covered = checked && !requirements.empty? && (requirements - verification[:checks].map { |c| c[:criterion] }).empty?
  vv = if checked && verification[:checks].any? { |c| c[:passed] == false }
         :fail
       elsif covered && verification[:checks].all? { |c| c[:passed] == true }
         :pass
       end
  runner = verification.is_a?(Hash) && verification[:runner_version] == 1
  if runner
    checks = Array(verification[:checks])
    requirements = Array(verification[:requirements])
    complete = !requirements.empty? && (requirements - checks.map { |c| c[:criterion] }).empty?
    vv = if checks.any? { |c| c[:passed] == false }
           :fail
         elsif complete && checked && checks.all? { |c| c[:passed] == true }
           :pass
         end
  end
  v[:verifier_verdict] = vv
  v[:confidence] = 1.0 if vv
  if vv == :fail || v.dig(:grounded, :verdict).to_s == 'refuted'
    score = [score || 0.0, 0.2].min
    success = false
  elsif vv == :pass
    score = [score || 0.0, 0.6].max
    success = true
  elsif (runner && vv.nil?) || score.nil? || source == 'error'
    success = nil
  elsif source.start_with?('heuristic')
    success = false
  elsif opts[:critic_pass] == false || v[:critic_pass] == false
    if score >= 0.6
      success = nil
    else
      score = [score, 0.3].min
      success = false
    end
  else
    success = score >= 0.6
  end
  known = !success.nil? && (!source.start_with?('heuristic') || !vv.nil?)
  verdict = if !known then :unknown
            elsif success then :solved
            elsif score >= 0.3 then :partial
            else :wrong
            end
  v.merge(score: score, success: success, verdict: verdict,
          critic_pass: opts.fetch(:critic_pass, v[:critic_pass]),
          training_score: known ? score : nil, decision_version: 1)
end

.run_verification(opts = {}) ⇒ Object

Trusted host-verifier API, deliberately NOT a model-facing tool. Execute an explicit host-owned contract before recording its result.

Raises:

• (ArgumentError)

# File 'lib/pwn/ai/agent/reward.rb', line 310

public_class_method def self.run_verification(opts = {})
  request = opts[:request].to_s
  sid = opts[:session_id].to_s
  rows = PWN::Sessions.load(session_id: sid)
  user = rows.reverse.find { |row| row[:role].to_s == 'user' }
  raise ArgumentError, 'verification must match the current session request' unless user && user[:content].to_s == request

  record = Verification.run(opts.fetch(:contract).merge(request: request)).merge(session_id: sid)
  PWN::Sessions.append(session_id: sid, role: 'verification', content: JSON.generate(record)) if opts.fetch(:commit, true)
  record
end

.scrub_preferences(opts = {}) ⇒ Object

P15 — one-shot ledger hygiene. Filters in place (rewrite jsonl) or report-only. Returns after:, dropped:, by_reason:, path:.

# File 'lib/pwn/ai/agent/reward.rb', line 1035

public_class_method def self.scrub_preferences(opts = {})
  dry = opts.key?(:dry_run) ? opts[:dry_run] : false
  path = PREFERENCES_FILE
  return { before: 0, after: 0, dropped: 0, dry_run: dry, path: path } unless File.exist?(path)

  raw = File.readlines(path)
  kept = []
  reasons = Hash.new(0)
  raw.each do |line|
    begin
      r = JSON.parse(line, symbolize_names: true)
    rescue StandardError
      reasons[:parse_error] += 1
      next
    end
    if usable_preference?(row: r)
      # P0 ops — backfill shape so trajectory_fraction is meaningful
      if r[:shape].to_s.empty?
        inferred = infer_shape(row: r)
        r = r.merge(shape: inferred) if inferred
      end
      kept << r
    else
      why = if r[:chosen].to_s.match?(/\A\s*CORRECTION:\s*/i)
              :correction_prose
            elsif r[:shape].to_s == 'fix_prose'
              :fix_prose
            elsif r[:chosen].to_s.length < (r[:rejected].to_s.length * 0.25)
              :chosen_too_short
            else
              :weak_geometry
            end
      reasons[why] += 1
    end
  end
  unless dry
    bak = "#{path}.bak-p15-#{Time.now.utc.strftime('%Y%m%d%H%M%S')}"
    FileUtils.cp(path, bak)
    File.open(path, 'w') { |f| kept.each { |r| f.puts(JSON.generate(r)) } }
  end
  {
    before: raw.length,
    after: kept.length,
    dropped: raw.length - kept.length,
    by_reason: reasons,
    dry_run: dry,
    path: path,
    backup: dry ? nil : bak
  }
rescue StandardError => e
  { error: "#{e.class}: #{e.message}" }
end

.semantic_ok(opts = {}) ⇒ Object

Supported Method Parameters

h = PWN::AI::Agent::Reward.semantic_ok( name: 'required - tool name', raw: 'required - JSON string returned by Dispatch.call', args: 'optional - the tool call arguments (used for BENIGN_EXIT)' )

Returns { ok:, semantic_ok:, exit:, err:, benign: }. :ok is the old proxy (handler didn't raise); :semantic_ok additionally knows that grep/diff/find exit≠0 with empty stderr is not a failure. Loop.run records Metrics on :ok but only records Mistakes on !semantic_ok.

# File 'lib/pwn/ai/agent/reward.rb', line 696

public_class_method def self.semantic_ok(opts = {})
  name = opts[:name].to_s
  raw  = opts[:raw].to_s
  ok   = raw.include?('"success":true')
  err  = raw[/"error":"([^"]{1,300})"/, 1]
  exit_code = raw[/"exit":(\d+)/, 1]&.to_i
  stderr    = raw[/"stderr":"([^"]{0,400})"/, 1].to_s

  benign = false
  shape  = nil
  if name == 'shell' && ok && exit_code && exit_code != 0
    cmd = extract_cmd(args: opts[:args])
    # 2.1 — ONLY BENIGN_EXIT regex × allowed codes. The old global
    # `stderr.empty? && exit==1 ⇒ benign` laundered real failures
    # (pipelines without pipefail, bare false, etc.) into "success".
    # For pipelines, match the LAST stage (post-pipe) first, then any.
    stages = cmd.split('|').map(&:strip)
    last   = stages.last.to_s
    benign = BENIGN_EXIT.any? { |rx, codes| last.match?(rx) && codes.include?(exit_code) }
    benign ||= stages.length > 1 && BENIGN_EXIT.any? { |rx, codes| stages.any? { |s| s.match?(rx) } && codes.include?(exit_code) && stderr.strip.empty? }
    shape = recoverable_shape(exit_code: exit_code, stderr: stderr, err: err)
  elsif !ok
    shape = recoverable_shape(exit_code: exit_code, stderr: stderr, err: err || raw[0, 200])
  end

  if raw.include?('invalid_payload') || err.to_s.include?('invalid_payload')
    semantic = false
    shape = :invalid_payload
    err ||= 'invalid_payload'
  elsif timeout_result?(err: err, raw: raw)
    semantic = false
    shape = :timeout
    err ||= raw[/timeout after \d+s/i] || 'timeout'
  else
    semantic = ok && (exit_code.nil? || exit_code.zero? || benign)
  end
  err ||= raw[/"stderr":"([^"]{4,300})"/, 1] unless semantic
  { ok: ok, semantic_ok: semantic, exit: exit_code, err: err, benign: benign, shape: shape }
end

.sentinel ⇒ Object

Supported Method Parameters

r = PWN::AI::Agent::Reward.sentinel

# File 'lib/pwn/ai/agent/reward.rb', line 504

public_class_method def self.sentinel
  s = normalize_sentinel(raw: load_sentinel)
  window = s[:window]
  n = window.length
  return { samples: n, status: :insufficient } if n < SENTINEL_WINDOW

  means = window_means(window: window)
  proxy = means[:proxy]
  judge = means[:judge]
  # Refuse to act on corrupt arithmetic — proxy must be a rate in [0,1].
  if proxy.nil? || proxy < 0.0 || proxy > 1.0
    return {
      samples: n,
      status: :corrupt_proxy,
      proxy: proxy,
      judge: judge&.round(3),
      reward_hacked: false,
      proxy_distrust: proxy_distrust
    }
  end

  human = 1.0 - user_correction_rate
  gap_pj = (proxy - judge).abs
  gap_ph = (proxy - human).abs
  hacked = gap_pj > SENTINEL_GAP || gap_ph > SENTINEL_GAP
  if hacked
    # 1.1 — freeze auto-Mistakes.record on tool:reward_signal after the
    # first open sig per gap-bucket. Endless ×13 fingerprints were
    # the loudest scar in every prompt and taught nothing. Open a
    # calibration path instead; park the sig as needs_code_change.
    bucket = "gap_pj=#{gap_pj.round(2)}|gap_ph=#{gap_ph.round(2)}"
    open_sig = defined?(Mistakes) ? Mistakes.for_tool(tool: 'reward_signal', unresolved_only: true) : []
    if open_sig.empty? && defined?(Mistakes)
      m = Mistakes.record(
        tool: 'reward_signal',
        error: "proxy success_rate #{proxy.round(2)} diverges from judge #{judge.round(2)} / human #{human.round(2)} by >#{SENTINEL_GAP}",
        source: :model,
        needs_code_change: true,
        meta: { bucket: bucket, proxy: proxy, judge: judge, human: human }
      )
      Mistakes.park(signature: m[:signature], reason: 'reward_signal needs calibration, not practice') if m && Mistakes.respond_to?(:park)
    end
    Curriculum.calibrate(predicted: proxy, actual: judge, engine: :reward_sentinel) if defined?(Curriculum) && Curriculum.respond_to?(:calibrate)
    # P4 — make sentinel ACTIONABLE: persist a distrust factor so
    # Metrics.to_context / Registry.rank haircut proxy success instead of
    # just opening another Mistakes row the model learns to ignore.
    set_proxy_distrust(gap: [gap_pj, gap_ph].max, proxy: proxy, judge: judge)
  else
    clear_proxy_distrust
  end
  {
    samples: n,
    proxy: proxy.round(3),
    judge: judge.round(3),
    human: human.round(3),
    gap_proxy_judge: gap_pj.round(3),
    gap_proxy_human: gap_ph.round(3),
    reward_hacked: hacked,
    proxy_distrust: proxy_distrust
  }
end

.set_proxy_distrust(opts = {}) ⇒ Object

# File 'lib/pwn/ai/agent/reward.rb', line 586

public_class_method def self.set_proxy_distrust(opts = {})
  s = normalize_sentinel(raw: load_sentinel)
  gap = opts[:gap].to_f
  proxy = opts[:proxy]
  # Guard: never set distrust from a nonsensical proxy (pre-ring-buffer
  # decay×to_i bug produced means ≫ 1.0 and hard-pegged distrust at 1.0).
  unless proxy.nil?
    pf = proxy.to_f
    return s[:proxy_distrust].to_f if pf < 0.0 || pf > 1.0
  end
  # Recalibrated: do NOT full-haircut raw success. 0.15→0.25, 0.30→0.50,
  # 0.45→0.70, hard cap 0.85 unless the gap is extreme (≥0.55 → 0.95).
  factor = ((((gap - SENTINEL_GAP) / SENTINEL_GAP) * 0.5) + 0.25).clamp(0.2, 0.85)
  s[:proxy_distrust] = factor
  s[:distrust_at] = Time.now.utc.iso8601
  s[:distrust_meta] = { proxy: opts[:proxy], judge: opts[:judge], gap: gap }
  FileUtils.mkdir_p(File.dirname(SENTINEL_FILE))
  atomic_write(path: SENTINEL_FILE, body: JSON.generate(s))
  factor
rescue StandardError
  nil
end

.usable_preference?(opts = {}) ⇒ Boolean

P15 — keep only usable preference pairs for balance/export/promote. Drops CORRECTION-only chosen, resolve rows without trajectory shape, and chosen≪rejected unless shape is a known trajectory form.

Returns:

• (Boolean)

# File 'lib/pwn/ai/agent/reward.rb', line 1009

public_class_method def self.usable_preference?(opts = {})
  r = opts.is_a?(Hash) && opts.key?(:row) ? opts[:row] : opts
  r = r.transform_keys(&:to_sym) if r.respond_to?(:transform_keys)
  chosen = r[:chosen].to_s
  rejected = r[:rejected].to_s
  shape = r[:shape].to_s
  source = r[:source].to_s
  return false if chosen.strip.empty? || rejected.strip.empty?
  return false if chosen.strip == rejected.strip
  return false if chosen.match?(/\A\s*CORRECTION:\s*/i) && chosen.length < 400
  return false if shape == 'fix_prose'
  # P25 — resolve rows must be trajectory-shaped to count as usable
  return false if source == 'mistakes_resolve' && !TRAJECTORY_SHAPES.include?(shape)

  # chosen ≪ rejected without trajectory shape → commentary, not policy
  unless TRAJECTORY_SHAPES.include?(shape)
    return false if rejected.length >= 200 && chosen.length < (rejected.length * 0.25) && chosen.length < 200
    return false if rejected.length >= 400 && chosen.length < 120
  end
  true
rescue StandardError
  false
end

.verify_as_reward(opts = {}) ⇒ Object

Supported Method Parameters

g = PWN::AI::Agent::Reward.verify_as_reward(final: text)

# File 'lib/pwn/ai/agent/reward.rb', line 768

public_class_method def self.verify_as_reward(opts = {})
  return nil unless defined?(Extrospection) && Extrospection.respond_to?(:verify)

  final = opts[:final].to_s
  claim = final[Learning::CLAIM_RX] if defined?(Learning)
  return nil if claim.to_s.empty?

  # P26 — drop metric crumbs ("cap 0.2") that match loose patterns
  if defined?(Learning) && Learning.respond_to?(:checkable_claim?, true)
    return nil unless Learning.send(:checkable_claim?, claim: claim)
  elsif claim.match?(/\A(?:cap|share|proxy|judge|success|only|now|gap|score|rate)\b/i) ||
        (claim.match?(/\d+\.\d+/) && !claim.match?(/\d+\.\d+\.\d+|CVE-/i))
    return nil
  end

  # 1.5 — sampled E3: always when flag true; never when false;
  # nil/auto → always on frontier, ~10% on local when CLAIM_RX hits.
  flag = agent_flag(key: :verify_as_reward, default: nil)
  eng  = (PWN::Env.dig(:ai, :active) if defined?(PWN::Env)).to_s.downcase
  local = eng == 'ollama'
  run = case flag
        when true then true
        when false then false
        else
          local ? (Digest::SHA256.hexdigest(claim.to_s)[0, 2].to_i(16) % 10).zero? : true
        end
  return nil unless run

  r = Extrospection.verify(claim: claim, commit: true)
  { claim: claim, verdict: r[:verdict], confidence: r[:confidence], reward: VERDICTS[r[:verdict]] || 0.5 }
rescue StandardError
  nil
end

.warm_sentinel(opts = {}) ⇒ Object

P10 — backfill the R3 ring from Learning outcomes so offline/local hosts reach SENTINEL_WINDOW without waiting for live remote introspect. Only fills empty slots; never flushes a warm window. Called by Curriculum.offline_judge and safe to cron.

# File 'lib/pwn/ai/agent/reward.rb', line 632

public_class_method def self.warm_sentinel(opts = {})
  s = normalize_sentinel(raw: load_sentinel)
  have = Array(s[:window]).length
  return { added: 0, samples: have, status: :full, proxy_distrust: proxy_distrust } if have >= SENTINEL_WINDOW
  return { added: 0, samples: have, status: :no_learning, proxy_distrust: proxy_distrust } unless defined?(Learning)

  need = SENTINEL_WINDOW - have
  limit = (opts[:limit] || [need * 4, 200].max).to_i
  # Prefer scored rows; fall back to success-boolean so local hosts still warm.
  rows = Learning.outcomes(limit: limit).select { |r| sentinel_outcome_known?(outcome: r) }
  scored, unscored = rows.partition { |r| !r[:score].nil? }
  ordered = scored.reverse + unscored.reverse
  added = 0
  ordered.each do |r|
    break if added >= need

    judge = if r[:score]
              r[:score].to_f.clamp(0.0, 1.0)
            else
              case r[:success]
              when true, 'true' then 0.75
              when 'soft' then 0.55
              when false, 'false' then 0.25
              else 0.5
              end
            end
    proxy = case r[:success]
            when true, 'true' then true
            when false, 'false', 'soft' then false
            else judge >= 0.6
            end
    record_sentinel(proxy: proxy, judge: judge)
    added += 1
  end
  final_n = Array(load_sentinel[:window]).length
  # Recompute distrust once window is full so controllers can engage.
  snap = final_n >= SENTINEL_WINDOW ? sentinel : { samples: final_n, status: :insufficient }
  {
    added: added,
    samples: final_n,
    status: (final_n >= SENTINEL_WINDOW ? :warmed_full : :warmed_partial),
    proxy_distrust: proxy_distrust,
    sentinel: snap.is_a?(Hash) ? snap.slice(:samples, :status, :reward_hacked, :proxy_distrust, :proxy, :judge) : nil
  }
rescue StandardError => e
  { added: 0, error: "#{e.class}: #{e.message}" }
end

.write_source_quota(opts = {}) ⇒ Object

Share of source among the newest WRITE_SOURCE_WINDOW prefs.

# File 'lib/pwn/ai/agent/reward.rb', line 894

public_class_method def self.write_source_quota(opts = {})
  source = opts[:source].to_s
  recent = preferences(limit: WRITE_SOURCE_WINDOW)
  return { over_cap: false, share: 0.0, n: 0, window: recent.length, underfilled: true } if recent.length < 10

  n = recent.count { |r| r[:source].to_s == source }
  share = n.to_f / recent.length
  target = TARGET_SOURCE_MIX[source]
  target_cap = target ? [WRITE_SOURCE_CAP, target + 0.05].min : WRITE_SOURCE_CAP
  {
    over_cap: share > target_cap,
    share: share.round(3),
    n: n,
    window: recent.length,
    source: source,
    cap: WRITE_SOURCE_CAP,
    target: target,
    underfilled: target ? share < (target * 0.5) : share < 0.05,
    deficit: target ? (target - share).round(3) : nil
  }
rescue StandardError
  { over_cap: false, share: 0.0, underfilled: true }
end