Module: CodexNotify::StreamProcessor

Defined in:
lib/codex_notify/stream_processor.rb

Class Method Summary collapse

Class Method Details

.drain_succeeded?(publisher) ⇒ Boolean

Returns:

  • (Boolean) 


136
137
138
139
 
# File 'lib/codex_notify/stream_processor.rb', line 136

def drain_succeeded?(publisher)
  result = publisher.drain
  result.failed.empty? && result.needs_review.empty?
end

.process_codex_log_stream(stream, token:, channel:, root_message:, initial_prompt: nil, user_name: 'user', include_tools: false, throttle_sec: 0.0, post_func:, sleep_func: Kernel.method(:sleep), publisher: nil) ⇒ Object 



9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/codex_notify/stream_processor.rb', line 9

def process_codex_log_stream(stream, token:, channel:, root_message:, initial_prompt: nil, user_name: 'user', include_tools: false,
                             throttle_sec: 0.0, post_func:, sleep_func: Kernel.method(:sleep), publisher: nil)
  if publisher
    return process_durable_stream(
      stream, root_message:, initial_prompt:, user_name:, include_tools:, publisher:
    )
  end

  last_sent_fingerprint = nil
  thread_ts_by_session = {}

  post_message = lambda do |message, thread_ts|
    CodexNotify::MessageFormatter.chunks(message).each do |part|
      post_func.call(token, channel, part, thread_ts)
      sleep_func.call(throttle_sec)
    end
  end

  post_root = lambda do |message|
    parts = CodexNotify::MessageFormatter.chunks(message).to_a
    response = post_func.call(token, channel, parts.shift, nil)
    thread_ts = response.fetch('ts').to_s
    sleep_func.call(throttle_sec)
    parts.each do |part|
      post_func.call(token, channel, part, thread_ts)
      sleep_func.call(throttle_sec)
    end
    thread_ts
  end

  post_root.call(root_message)

  unless initial_prompt.nil? || initial_prompt.empty?
    message = CodexNotify::MessageFormatter.message(title: user_name, body: initial_prompt, presentation: :plain)
    thread_ts_by_session['__default__'] = post_root.call(message)
  end

  stream.each do |raw|
    line = raw.strip
    next if line.empty?

    event = JSON.parse(line)
    session_id = CodexNotify::LogEventParser.extract_session_id(event) || '__default__'
    extracted_events = CodexNotify::LogEventParser.extract_events(event)
    next if extracted_events.empty?

    extracted_events.each do |kind, text, part_type|
      next if text.empty?
      next if part_type == 'tool' && !include_tools

      fingerprint = "#{session_id}:#{part_type}:#{kind}:#{text[0, 240]}"
      next if fingerprint == last_sent_fingerprint

      last_sent_fingerprint = fingerprint
      title = kind == 'assistant' ? 'assistant' : kind
      thread_ts = thread_ts_by_session[session_id]
      presentation = %w[user assistant system].include?(title) ? :plain : :block
      message_title = kind == 'user' ? user_name : title
      message = CodexNotify::MessageFormatter.message(title: message_title, body: text, presentation:)

      if kind == 'user'
        if thread_ts
          post_message.call(message, thread_ts)
        else
          thread_ts = post_root.call(message)
          thread_ts_by_session[session_id] = thread_ts
        end
        next
      end

      post_message.call(message, thread_ts)
    end
  rescue JSON::ParserError
    next
  end

  0
end

.process_durable_stream(stream, root_message:, initial_prompt:, user_name:, include_tools:, publisher:) ⇒ Object 



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
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
 
# File 'lib/codex_notify/stream_processor.rb', line 88

def process_durable_stream(stream, root_message:, initial_prompt:, user_name:, include_tools:, publisher:)
  last_sent_fingerprint = nil
  monitor_key = publisher.key('log-monitor', Process.pid)
  publisher.publish_standalone(key: monitor_key, message: root_message)
  return 1 unless drain_succeeded?(publisher)

  unless initial_prompt.nil? || initial_prompt.empty?
    message = CodexNotify::MessageFormatter.message(title: user_name, body: initial_prompt, presentation: :plain)
    publisher.publish_root_or_reply(key: publisher.key('log', '__default__'), message:)
    return 1 unless drain_succeeded?(publisher)
  end

  stream.each do |raw|
    line = raw.strip
    next if line.empty?

    event = JSON.parse(line)
    session_id = CodexNotify::LogEventParser.extract_session_id(event) || '__default__'
    extracted_events = CodexNotify::LogEventParser.extract_events(event)
    next if extracted_events.empty?

    extracted_events.each do |kind, body, part_type|
      next if body.empty?
      next if part_type == 'tool' && !include_tools

      fingerprint = "#{session_id}:#{part_type}:#{kind}:#{body[0, 240]}"
      next if fingerprint == last_sent_fingerprint

      last_sent_fingerprint = fingerprint
      title = kind == 'assistant' ? 'assistant' : kind
      presentation = %w[user assistant system].include?(title) ? :plain : :block
      message_title = kind == 'user' ? user_name : title
      message = CodexNotify::MessageFormatter.message(title: message_title, body:, presentation:)
      key = publisher.key('log', session_id)
      if kind == 'user'
        publisher.publish_root_or_reply(key:, message:)
      else
        publisher.publish_reply(key:, message:, recovery_root_message: message)
      end
      return 1 unless drain_succeeded?(publisher)
    end
  rescue JSON::ParserError
    next
  end

  0
end