Class: Harnex::Adapters::Base

Inherits:
Object
  • Object
show all
Defined in:
lib/harnex/adapters/base.rb

Direct Known Subclasses

Claude, Codex, CodexAppServer, Generic, Opencode, Pi

Constant Summary collapse

PROMPT_PREFIXES = 
[">", "\u203A", "\u276F"].freeze
AGENT_VERSION_TIMEOUT_SECONDS = 
2.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(key, extra_args = []) ⇒ Base

Returns a new instance of Base.



22
23
24
25
 
# File 'lib/harnex/adapters/base.rb', line 22

def initialize(key, extra_args = [])
  @key = key
  @extra_args = extra_args.dup
end

Instance Attribute Details

#key ⇒ Object (readonly)

Adapter contract — subclasses MUST implement:

base_command          -> Array[String]  CLI args to spawn

Subclasses MAY override:

input_state(text)     -> Hash           Parse screen for state
build_send_payload    -> Hash           Build injection payload
inject_exit(writer)   -> void           Send a stop/exit sequence
infer_repo_path(argv) -> String         Extract repo path from CLI args
wait_for_sendable     -> String         Wait for a send-ready snapshot


20
21
22
 
# File 'lib/harnex/adapters/base.rb', line 20

def key
  @key
end

Instance Method Details

#agent_version ⇒ Object

Probes <base_command.first> --version with a short timeout and memoizes the result for the adapter's lifetime. Returns nil when the binary is missing, exits non-zero, or stalls past the timeout.



53
54
55
56
57
 
# File 'lib/harnex/adapters/base.rb', line 53

def agent_version
  return @agent_version if defined?(@agent_version)

  @agent_version = probe_agent_version
end

#base_command ⇒ Object

Raises:

  • (NotImplementedError) 


67
68
69
 
# File 'lib/harnex/adapters/base.rb', line 67

def base_command
  raise NotImplementedError, "#{self.class} must define #base_command"
end

#build_command ⇒ Object 



63
64
65
 
# File 'lib/harnex/adapters/base.rb', line 63

def build_command
  base_command + @extra_args
end

#build_send_payload(text:, submit:, enter_only:, screen_text:, force: false) ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/harnex/adapters/base.rb', line 130

def build_send_payload(text:, submit:, enter_only:, screen_text:, force: false)
  state = input_state(screen_text)
  if !force && blocked_state?(state, enter_only: enter_only)
    raise ArgumentError, blocked_message(state, enter_only: enter_only)
  end

  payload = enter_only ? "" : text.to_s
  payload << submit_bytes if submit || enter_only

  {
    text: payload,
    newline: false,
    input_state: state,
    force: force
  }
end

#context_telemetry_source ⇒ Object 



98
99
100
 
# File 'lib/harnex/adapters/base.rb', line 98

def context_telemetry_source
  nil
end

#context_telemetry_supported? ⇒ Boolean

Whether this adapter exposes active context-window occupancy separately from cumulative usage. Supported sources can still yield no valid sample.

Returns:

  • (Boolean) 


94
95
96
 
# File 'lib/harnex/adapters/base.rb', line 94

def context_telemetry_supported?
  false
end

#describe ⇒ Object 



59
60
61
 
# File 'lib/harnex/adapters/base.rb', line 59

def describe
  { transport: transport }
end

#infer_repo_path(_argv) ⇒ Object 



71
72
73
 
# File 'lib/harnex/adapters/base.rb', line 71

def infer_repo_path(_argv)
  Dir.pwd
end

#inject_exit(writer, delay_ms: 0) ⇒ Object 



147
148
149
150
151
152
153
 
# File 'lib/harnex/adapters/base.rb', line 147

def inject_exit(writer, delay_ms: 0)
  writer.write("/exit")
  writer.flush
  sleep(delay_ms / 1000.0) if delay_ms.positive?
  writer.write(submit_bytes)
  writer.flush
end

#input_state(screen_text) ⇒ Object 



75
76
77
78
79
80
 
# File 'lib/harnex/adapters/base.rb', line 75

def input_state(screen_text)
  {
    state: "unknown",
    input_ready: nil
  }
end

#parse_session_summary(_transcript_tail) ⇒ Object 



82
83
84
 
# File 'lib/harnex/adapters/base.rb', line 82

def parse_session_summary(_transcript_tail)
  {}
end

#provider ⇒ Object

Vendor of the underlying agent — populates DISPATCH meta.agent_provider. Subclasses override (claude → "anthropic", codex → "openai").



36
37
38
 
# File 'lib/harnex/adapters/base.rb', line 36

def provider
  nil
end

#send_wait_seconds(submit:, enter_only:) ⇒ Object 



102
103
104
 
# File 'lib/harnex/adapters/base.rb', line 102

def send_wait_seconds(submit:, enter_only:)
  0.0
end

#transport ⇒ Object

Default transport. Structured adapters override to :stdio_jsonrpc (Codex app-server) or :stdio_jsonl_rpc (Pi RPC); Session#run uses this to pick the I/O path.



30
31
32
 
# File 'lib/harnex/adapters/base.rb', line 30

def transport
  :pty
end

#usage_input_includes_cached? ⇒ Boolean

Whether the usage this adapter captures reports input_tokens INCLUSIVE of cached_tokens. Depends on the capture path, not just the provider: codex app-server JSON reports cached as a subset of input, while the codex TUI line the PTY adapter scrapes shows non-cached input with cached as a separate additive count. Feeds Pricing.compute (plan 33 Phase 2).

Returns:

  • (Boolean) 


46
47
48
 
# File 'lib/harnex/adapters/base.rb', line 46

def usage_input_includes_cached?
  false
end

#usage_telemetry_supported? ⇒ Boolean

Whether this adapter has a supported source for token/cost telemetry. A supported source can still be missing for a particular run.

Returns:

  • (Boolean) 


88
89
90
 
# File 'lib/harnex/adapters/base.rb', line 88

def usage_telemetry_supported?
  false
end

#wait_for_sendable(screen_snapshot_fn, submit:, enter_only:, force:) ⇒ Object 



110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/harnex/adapters/base.rb', line 110

def wait_for_sendable(screen_snapshot_fn, submit:, enter_only:, force:)
  snapshot = screen_snapshot_fn.call
  return snapshot if force

  wait_secs = send_wait_seconds(submit: submit, enter_only: enter_only).to_f
  return snapshot unless wait_secs.positive?

  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + wait_secs
  state = input_state(snapshot)

  while Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline &&
        wait_for_sendable_state?(state, submit: submit, enter_only: enter_only)
    sleep 0.05
    snapshot = screen_snapshot_fn.call
    state = input_state(snapshot)
  end

  snapshot
end

#wait_for_sendable_state?(_state, submit:, enter_only:) ⇒ Boolean

Returns:

  • (Boolean) 


106
107
108
 
# File 'lib/harnex/adapters/base.rb', line 106

def wait_for_sendable_state?(_state, submit:, enter_only:)
  false
end