Module: PWN::AI::Agent::PromptCache

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

Overview

Hermes-style prompt-cache breakpoints. The static system prefix + SKILLS index stay cacheable; dynamic MEMORY / LEARNING / MISTAKES / METRICS / EXTROSPECTION stay uncached so a turn-local write is not baked into the prefix.

Anthropic: system is an array of text blocks; the last static block carries cache_control: { type: 'ephemeral' }. OpenAI Chat Completions: two system/developer messages (static then dynamic) plus prompt_cache_key for cache routing. xAI Grok Chat Completions: the same message split plus the x-grok-conv-id request header (sticky server routing). Gemini generateContent: systemInstruction.parts = [static, dynamic] so implicit prefix cache can hit the stable head. Ollama: no-op.

Constant Summary collapse

STATIC_HEAD = 
%w[ENVIRONMENT].freeze
DYNAMIC_HEADS = 
[
  'MEMORY',
  'RECENT TURNS',
  'LEARNING',
  'OPERATOR INBOX',
  'KNOWN MISTAKES',
  'KNOWN FIXES',
  'TOOL EFFECTIVENESS',
  'POLICY',
  'EXTROSPECTION',
  'TOOL USE'
].freeze

Class Method Summary collapse

Class Method Details

.anthropic_system_blocks(opts = {}) ⇒ Object

Supported Method Parameters

blocks = PWN::AI::Agent::PromptCache.anthropic_system_blocks(text: system_prompt)

→ [ text: static, cache_control:{type:'ephemeral'}, text: dynamic ] (empty blocks omitted)



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 78

public_class_method def self.anthropic_system_blocks(opts = {})
  parts = split_system(text: opts[:text])
  blocks = []
  unless parts[:static].to_s.strip.empty?
    blocks << {
      type: 'text',
      text: parts[:static],
      cache_control: { type: 'ephemeral' }
    }
  end
  blocks << { type: 'text', text: parts[:dynamic] } unless parts[:dynamic].to_s.strip.empty?
  blocks
rescue StandardError
  txt = opts[:text].to_s
  return [] if txt.strip.empty?

  [{ type: 'text', text: txt }]
end

.authors ⇒ Object 



195
196
197
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 195

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

.cache_key(opts = {}) ⇒ Object

Supported Method Parameters

key = PWN::AI::Agent::PromptCache.cache_key(text: system_prompt) key = PWN::AI::Agent::PromptCache.cache_key(key: 'explicit-override')

Stable OpenAI prompt_cache_key / xAI x-grok-conv-id. Hashes the static prefix so MEMORY / LEARNING churn does not rotate the key.



166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 166

public_class_method def self.cache_key(opts = {})
  explicit = opts[:key]
  return explicit.to_s unless explicit.to_s.strip.empty?

  raw = opts[:text].to_s
  static = raw.empty? ? '' : split_system(text: raw)[:static].to_s
  static = raw if static.strip.empty?
  static = 'pwn-ai' if static.strip.empty?
  "pwn-#{Digest::SHA256.hexdigest(static)[0, 24]}"
rescue StandardError
  'pwn-ai'
end

.cache_marks(opts = {}) ⇒ Object

Supported Method Parameters

n = PWN::AI::Agent::PromptCache.cache_marks(blocks: [...])



182
183
184
185
186
187
188
189
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 182

public_class_method def self.cache_marks(opts = {})
  Array(opts[:blocks]).count do |b|
    next false unless b.is_a?(Hash)

    cc = b[:cache_control] || b['cache_control']
    cc.is_a?(Hash) && (cc[:type] || cc['type']).to_s == 'ephemeral'
  end
end

.enabled?(opts = {}) ⇒ Boolean

Supported Method Parameters

ok = PWN::AI::Agent::PromptCache.enabled?(engine: :anthropic)

Returns:

  • (Boolean) 


40
41
42
43
44
45
46
47
48
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 40

public_class_method def self.enabled?(opts = {})
  flag = agent_flag(key: :prompt_cache, default: true)
  return false if flag == false || flag.to_s.match?(/\A(0|false|no|off)\z/i)

  engine = (opts[:engine] || active_engine).to_s.downcase.to_sym
  supports.include?(engine)
rescue StandardError
  false
end

.gemini_system_instruction(opts = {}) ⇒ Object

Supported Method Parameters

inst = PWN::AI::Agent::PromptCache.gemini_system_instruction(text: system_prompt)

→ { parts: [static, dynamic] } (empty omitted)



144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 144

public_class_method def self.gemini_system_instruction(opts = {})
  parts = split_system(text: opts[:text])
  out = []
  out << { text: parts[:static] } unless parts[:static].to_s.strip.empty?
  out << { text: parts[:dynamic] } unless parts[:dynamic].to_s.strip.empty?
  return nil if out.empty?

  { parts: out }
rescue StandardError
  txt = opts[:text].to_s
  return nil if txt.strip.empty?

  { parts: [{ text: txt }] }
end

.help ⇒ Object 



199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 199

public_class_method def self.help
  puts <<~USAGE
    USAGE:
      parts  = PWN::AI::Agent::PromptCache.split_system(text: system_prompt)
      blocks = PWN::AI::Agent::PromptCache.anthropic_system_blocks(text: system_prompt)
      msgs   = PWN::AI::Agent::PromptCache.openai_messages(messages: [...])
      inst   = PWN::AI::Agent::PromptCache.gemini_system_instruction(text: system_prompt)
      key    = PWN::AI::Agent::PromptCache.cache_key(text: system_prompt)
      # Engine chat_with_tools uses the helpers automatically when
      # PWN::Env[:ai][:agent][:prompt_cache] is truthy (default).

      Toggle via PWN::Env[:ai][:agent][:prompt_cache]
      #{self}.authors
  USAGE
end

.openai_messages(opts = {}) ⇒ Object

Supported Method Parameters

messages = PWN::AI::Agent::PromptCache.openai_messages(messages: [...])

Split the first system/developer message into a stable static prefix and a turn-local dynamic tail so OpenAI / xAI implicit prefix matching can hit across turns. Does NOT emit Anthropic cache_control.



105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 105

public_class_method def self.openai_messages(opts = {})
  messages = Array(opts[:messages]).map { |m| m.is_a?(Hash) ? m.dup : m }
  idx = messages.index do |m|
    next false unless m.is_a?(Hash)

    %w[system developer].include?((m[:role] || m['role']).to_s)
  end
  return messages unless idx

  msg = messages[idx]
  role = msg[:role] || msg['role']
  text = (msg[:content] || msg['content']).to_s
  parts = split_system(text: text)
  return messages if parts[:static].to_s.strip.empty? || parts[:dynamic].to_s.strip.empty?

  static_msg = msg.dup
  if static_msg.key?(:content) || !static_msg.key?('content')
    static_msg[:content] = parts[:static]
  else
    static_msg['content'] = parts[:static]
  end

  dynamic_msg = if msg.key?(:role) || !msg.key?('role')
                  { role: role, content: parts[:dynamic] }
                else
                  { 'role' => role, 'content' => parts[:dynamic] }
                end

  messages[idx, 1] = [static_msg, dynamic_msg]
  messages
rescue StandardError
  Array(opts[:messages])
end

.split_system(opts = {}) ⇒ Object

Supported Method Parameters

parts = PWN::AI::Agent::PromptCache.split_system(text: system_prompt)

→ { static: String, dynamic: String, skills: String }



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 55

public_class_method def self.split_system(opts = {})
  text = opts[:text].to_s
  return { static: '', dynamic: '', skills: '' } if text.empty?

  dyn_idx = first_dynamic_index(text: text)
  if dyn_idx.nil?
    skills = extract_skills(text: text)
    return { static: text, dynamic: '', skills: skills }
  end

  static = text[0...dyn_idx].rstrip
  dynamic = text[dyn_idx..].to_s
  { static: static, dynamic: dynamic, skills: extract_skills(text: static) }
rescue StandardError
  { static: opts[:text].to_s, dynamic: '', skills: '' }
end

.supports ⇒ Object 



191
192
193
 
# File 'lib/pwn/ai/agent/prompt_cache.rb', line 191

public_class_method def self.supports
  %i[anthropic openai grok gemini]
end