Module: PWN::Sessions

Defined in:
lib/pwn/sessions.rb

Overview

PWN::Sessions provides session management for pwn-ai (and other drivers) equivalent to Hermes sessions (list, resume, transcripts, stats). Sessions are stored as JSONL transcripts in ~/.pwn/sessions/ for durability and easy search/append. pwn-ai agent mode auto-creates and appends to a session on each activation.

Constant Summary collapse

SESSIONS_DIR = 
File.join(Dir.home, '.pwn', 'sessions')

Class Method Summary collapse

Class Method Details

.append(opts = {}) ⇒ Object

Supported Method Parameters

PWN::Sessions.append(

session_id: 'required',
role: 'user|assistant|system|observation',
content: 'the message or obs'

)



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

public_class_method def self.append(opts = {})
  sid = opts[:session_id]
  raise 'ERROR: session_id required' unless sid

  path = File.join(sessions_dir, "#{sid}.jsonl")
  raise "Session #{sid} not found" unless File.exist?(path)

  entry = {
    role: opts[:role] || 'user',
    content: opts[:content],
    timestamp: Time.now.utc.iso8601
  }
  File.open(path, 'a') { |f| f.puts(JSON.dump(entry)) }
  entry
end

.authors ⇒ Object

Author(s)

0day Inc. <support@0dayinc.com>



140
141
142
 
# File 'lib/pwn/sessions.rb', line 140

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

.create(opts = {}) ⇒ Object

Supported Method Parameters

session = PWN::Sessions.create(

title: 'optional - human title',
source: 'optional - e.g. pwn-ai-repl'

)



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/pwn/sessions.rb', line 44

public_class_method def self.create(opts = {})
  dir = sessions_dir
  ts = Time.now.utc.strftime('%Y%m%d_%H%M%S')
  rand = SecureRandom.hex(4)
  id = "#{ts}_#{rand}"
  path = File.join(dir, "#{id}.jsonl")

  meta = {
    id: id,
    title: opts[:title] || "pwn-ai session #{id}",
    source: opts[:source] || 'pwn-ai',
    created_at: Time.now.utc.iso8601
  }

  File.open(path, 'w') do |f|
    f.puts(JSON.dump(role: 'system', content: "Session started: #{meta[:title]}", timestamp: meta[:created_at]))
  end
  { id: id, path: path, meta: meta }
end

.delete(opts = {}) ⇒ Object

rubocop:disable Naming/PredicateMethod



119
120
121
122
123
124
 
# File 'lib/pwn/sessions.rb', line 119

public_class_method def self.delete(opts = {}) # rubocop:disable Naming/PredicateMethod
  sid = opts[:session_id]
  path = File.join(sessions_dir, "#{sid}.jsonl")
  FileUtils.rm_f(path)
  true
end

.help ⇒ Object

Display Usage for this Module



145
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/pwn/sessions.rb', line 145

public_class_method def self.help
  puts <<~USAGE
    USAGE:
      sess = PWN::Sessions.create(title: 'recon on target.com')
      PWN::Sessions.append(session_id: sess[:id], role: 'user', content: 'Run NmapIt...')
      transcript = PWN::Sessions.load(session_id: sess[:id])
      hist = PWN::Sessions.to_response_history(session_id: sess[:id])
      PWN::Sessions.list
      PWN::Sessions.stats
      PWN::Sessions.delete(session_id: sess[:id])

      #{self}.authors
  USAGE
end

.list ⇒ Object

Supported Method Parameters

sessions = PWN::Sessions.list



26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/pwn/sessions.rb', line 26

public_class_method def self.list
  dir = sessions_dir
  Dir.glob(File.join(dir, '*.jsonl')).reverse.map do |f|
    {
      id: File.basename(f, '.jsonl'),
      path: f,
      size: File.size(f),
      mtime: File.mtime(f).utc.iso8601,
      lines: File.readlines(f).count
    }
  end
end

.load(opts = {}) ⇒ Object

Supported Method Parameters

transcript = PWN::Sessions.load(session_id: ‘required’)



88
89
90
91
92
93
94
 
# File 'lib/pwn/sessions.rb', line 88

public_class_method def self.load(opts = {})
  sid = opts[:session_id]
  path = File.join(sessions_dir, "#{sid}.jsonl")
  return [] unless File.exist?(path)

  File.readlines(path).map { |l| JSON.parse(l, symbolize_names: true) }
end

.sessions_dir ⇒ Object

Supported Method Parameters

dir = PWN::Sessions.sessions_dir



19
20
21
22
 
# File 'lib/pwn/sessions.rb', line 19

public_class_method def self.sessions_dir
  FileUtils.mkdir_p(SESSIONS_DIR)
  SESSIONS_DIR
end

.stats ⇒ Object

Supported Method Parameters

stats = PWN::Sessions.stats



128
129
130
131
132
133
134
135
136
 
# File 'lib/pwn/sessions.rb', line 128

public_class_method def self.stats
  sessions = list
  {
    total_sessions: sessions.size,
    total_lines: sessions.sum { |s| s[:lines] },
    oldest: sessions.last ? sessions.last[:mtime] : nil,
    newest: sessions.first ? sessions.first[:mtime] : nil
  }
end

.to_response_history(opts = {}) ⇒ Object

Supported Method Parameters

history_for_ai = PWN::Sessions.to_response_history(session_id:) (converts transcript to the response_history format used by PWN::AI::* .chat)



99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/pwn/sessions.rb', line 99

public_class_method def self.to_response_history(opts = {})
  transcript = load(session_id: opts[:session_id])
  choices = transcript.map do |e|
    {
      role: e[:role],
      content: e[:content]
    }
  end

  {
    id: opts[:session_id],
    object: 'session.transcript',
    model: 'pwn-ai',
    usage: {},
    choices: choices
  }
end