use std::collections::HashMap;
use std::io::{IsTerminal, Read, Write};
use std::path::PathBuf;
use std::process::Command;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use hh_core::adapter::{Adapter, AdapterContext, AdapterHandle, AdapterOutcome};
use hh_core::blob::BlobStore;
use hh_core::event::{AdapterStatus, Event, EventKind, NewSession};
use hh_core::store::{EventWriter, Store};
use portable_pty::{native_pty_system, CommandBuilder, PtySize};
use signal_hook::consts::{SIGINT, SIGTERM};
#[cfg(unix)]
use signal_hook::consts::SIGWINCH;
use signal_hook::flag::register as register_flag;
use crate::watcher::{spawn_watcher, WatchOptions};
const CHUNK_BYTES: usize = 8 * 1024;
const CHUNK_INTERVAL: Duration = Duration::from_millis(50);
#[derive(Debug, Clone)]
pub struct RunOptions {
pub command: Vec<String>,
pub adapter: Option<String>,
pub cwd: PathBuf,
pub max_file_size: u64,
pub record_input: bool,
pub record_binary: bool,
pub extra_ignore: Vec<String>,
pub internal_exclude: Vec<PathBuf>,
pub hh_version: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunOutcome {
pub session_id: String,
pub short_id: String,
pub exit_code: Option<i32>,
pub duration_ms: i64,
pub event_count: i64,
pub steps: i64,
pub files_changed: i64,
pub status: &'static str,
}
struct RawModeGuard;
impl Drop for RawModeGuard {
fn drop(&mut self) {
let _ = crossterm::terminal::disable_raw_mode();
}
}
#[allow(clippy::too_many_lines)] pub fn run(store: &Store, opts: &RunOptions) -> crate::Result<RunOutcome> {
let start = Instant::now();
let started_at = now_unix_ms();
let session_uuid = hh_core::event::now_v7();
let adapter: Option<Box<dyn Adapter>> = match opts.adapter.as_deref() {
Some(name) => match hh_core::resolve_adapter_override(name) {
Some(a) => Some(a),
None => {
return Err(crate::RecordError::Adapter(format!(
"unknown adapter `{name}` (available: claude-code)"
)));
}
},
None => hh_core::select_adapter(&opts.command, &opts.cwd),
};
let agent_kind = match adapter.as_ref() {
Some(a) => a.agent_kind(),
None => crate::agent::detect_agent(&opts.command),
};
let adapter_status = if adapter.is_some() {
AdapterStatus::Active
} else {
AdapterStatus::None
};
let git = crate::git::GitMeta::capture(&opts.cwd);
let new_session = NewSession {
id: session_uuid,
started_at,
agent_kind,
adapter_status,
command: opts.command.clone(),
cwd: opts.cwd.clone(),
hostname: hostname(),
hh_version: opts.hh_version.clone(),
model: None,
git_branch: git.branch,
git_sha: git.sha,
git_dirty: git.dirty,
};
let created = store.create_session(&new_session)?;
let session_id = created.id;
let short_id = created.short_id;
let writer = Arc::new(Mutex::new(store.event_writer()?));
let watch_opts = WatchOptions {
cwd: opts.cwd.clone(),
max_file_size: opts.max_file_size,
record_binary: opts.record_binary,
extra_ignore: opts.extra_ignore.clone(),
internal_exclude: opts.internal_exclude.clone(),
};
let blobs = Arc::new(BlobStore::new(store.blobs().root().to_path_buf()));
let watcher = spawn_watcher(
watch_opts,
Arc::clone(&writer),
Arc::clone(&blobs),
session_id.clone(),
start,
)?;
let adapter_stop = Arc::new(AtomicBool::new(false));
let (drain_thread, adapter_outcome_handle) = match adapter {
Some(adapter) => {
let ctx = AdapterContext {
session_id: session_id.clone(),
started_at_unix_ms: started_at,
cwd: opts.cwd.clone(),
command: opts.command.clone(),
blobs: Arc::clone(&blobs),
stop: Arc::clone(&adapter_stop),
projects_dir: None,
};
let AdapterHandle { events, outcome } = adapter
.spawn(ctx)
.map_err(|e| crate::RecordError::Pty(format!("spawn claude adapter: {e}")))?;
let writer_for_drain = Arc::clone(&writer);
let drain = std::thread::Builder::new()
.name("hh-adapter-drain".into())
.spawn(move || run_adapter_drain(events, writer_for_drain))
.map_err(|e| crate::RecordError::Pty(format!("spawn adapter drain: {e}")))?;
(Some(drain), Some(outcome))
}
None => (None, None),
};
let pty_system = native_pty_system();
let initial_size = current_pty_size();
let pty_pair = pty_system
.openpty(initial_size)
.map_err(|e| crate::RecordError::Pty(e.to_string()))?;
let mut cmd = CommandBuilder::new(&opts.command[0]);
for arg in &opts.command[1..] {
cmd.arg(arg);
}
cmd.cwd(&opts.cwd);
cmd.env("HH_SESSION_ID", &session_id);
let child = pty_pair
.slave
.spawn_command(cmd)
.map_err(|e| crate::RecordError::Spawn {
command: opts.command.join(" "),
reason: e.to_string(),
})?;
drop(pty_pair.slave);
let reader = pty_pair
.master
.try_clone_reader()
.map_err(|e| crate::RecordError::Pty(e.to_string()))?;
let stdin_writer = pty_pair
.master
.take_writer()
.map_err(|e| crate::RecordError::Pty(e.to_string()))?;
let master = pty_pair.master;
let raw_guard = if std::io::stdin().is_terminal() {
match crossterm::terminal::enable_raw_mode() {
Ok(()) => Some(RawModeGuard),
Err(e) => {
eprintln!("hh: warning: could not enter raw mode: {e}");
None
}
}
} else {
None
};
let stop = Arc::new(AtomicBool::new(false));
let resize_flag = Arc::new(AtomicBool::new(false));
#[cfg(unix)]
let _ = register_flag(SIGWINCH, Arc::clone(&resize_flag));
let _ = register_flag(SIGTERM, Arc::clone(&stop));
let _ = register_flag(SIGINT, Arc::clone(&stop));
let writer_for_reader = Arc::clone(&writer);
let blobs_for_reader = Arc::clone(&blobs);
let session_id_for_reader = session_id.clone();
let reader_stop = Arc::new(AtomicBool::new(false));
let reader_stop_for_check = Arc::clone(&reader_stop);
let reader_thread = std::thread::Builder::new()
.name("hh-pty-reader".into())
.spawn(move || {
run_reader(
reader,
writer_for_reader,
blobs_for_reader,
session_id_for_reader,
start,
reader_stop_for_check,
);
})
.map_err(|e| crate::RecordError::Pty(format!("spawn reader thread: {e}")))?;
let writer_for_stdin = Arc::clone(&writer);
let session_id_for_stdin = session_id.clone();
let record_input = opts.record_input;
let stdin_thread = std::thread::Builder::new()
.name("hh-stdin-proxy".into())
.spawn(move || {
run_stdin_proxy(
stdin_writer,
writer_for_stdin,
session_id_for_stdin,
start,
record_input,
);
})
.map_err(|e| crate::RecordError::Pty(format!("spawn stdin thread: {e}")))?;
let mut child = child;
let mut killed_by_signal = false;
let exit_status = loop {
if stop.load(Ordering::Acquire) {
killed_by_signal = true;
let _ = child.kill();
}
if resize_flag.swap(false, Ordering::AcqRel) {
let size = current_pty_size();
let _ = master.resize(size);
}
match child
.try_wait()
.map_err(|e| crate::RecordError::Child(e.to_string()))?
{
Some(status) => break status,
None => std::thread::sleep(Duration::from_millis(10)),
}
};
reader_stop.store(true, Ordering::Release);
drop(master);
let _ = reader_thread.join();
drop(stdin_thread);
watcher.stop_and_join();
adapter_stop.store(true, Ordering::Release);
if let Some(t) = drain_thread {
let _ = t.join();
}
let adapter_outcome = match adapter_outcome_handle {
Some(h) => Some(
h.join()
.map_err(|_| crate::RecordError::Pty("claude adapter thread panicked".into()))?,
),
None => None,
};
drop(raw_guard); let duration_ms = i64::try_from(start.elapsed().as_millis()).unwrap_or(i64::MAX);
let code = i32::try_from(exit_status.exit_code()).unwrap_or(i32::MAX);
let (exit_code, status) = if killed_by_signal {
(Some(code), "interrupted")
} else if exit_status.success() {
(Some(0), "ok")
} else {
(Some(code), "error")
};
let (event_count, steps, files_changed) = finalize(
store,
&session_id,
&writer,
adapter_outcome.as_ref(),
exit_code,
status,
)?;
Ok(RunOutcome {
session_id,
short_id,
exit_code,
duration_ms,
event_count,
steps,
files_changed,
status,
})
}
#[allow(clippy::needless_pass_by_value)]
fn run_reader(
mut reader: Box<dyn Read + Send>,
writer: Arc<Mutex<EventWriter>>,
blobs: Arc<BlobStore>,
session_id: String,
start: Instant,
stop: Arc<AtomicBool>,
) {
let mut out = std::io::stdout();
let mut buf = [0u8; CHUNK_BYTES];
let mut acc: Vec<u8> = Vec::with_capacity(CHUNK_BYTES);
let mut last_flush = Instant::now();
loop {
if stop.load(Ordering::Acquire) {
break;
}
match reader.read(&mut buf) {
Ok(0) => break, Ok(n) => {
let _ = out.write_all(&buf[..n]);
let _ = out.flush();
acc.extend_from_slice(&buf[..n]);
if acc.len() >= CHUNK_BYTES || last_flush.elapsed() >= CHUNK_INTERVAL {
flush_terminal_chunk(&writer, &blobs, &session_id, start, &acc);
acc.clear();
last_flush = Instant::now();
}
}
Err(e) => {
if e.kind() != std::io::ErrorKind::Interrupted {
break;
}
}
}
}
if !acc.is_empty() {
flush_terminal_chunk(&writer, &blobs, &session_id, start, &acc);
}
}
#[allow(clippy::needless_pass_by_value)] fn run_stdin_proxy(
mut stdin_writer: Box<dyn Write + Send>,
writer: Arc<Mutex<EventWriter>>,
session_id: String,
start: Instant,
record_input: bool,
) {
let mut stdin = std::io::stdin();
let mut buf = [0u8; CHUNK_BYTES];
loop {
match stdin.read(&mut buf) {
Ok(0) => break, Ok(n) => {
if record_input {
flush_input_chunk(&writer, &session_id, start, &buf[..n]);
}
if stdin_writer.write_all(&buf[..n]).is_err() {
break; }
let _ = stdin_writer.flush();
}
Err(e) => {
if e.kind() != std::io::ErrorKind::Interrupted {
break;
}
}
}
}
}
#[allow(clippy::needless_pass_by_value)]
fn run_adapter_drain(events: std::sync::mpsc::Receiver<Event>, writer: Arc<Mutex<EventWriter>>) {
let mut calls: HashMap<String, i64> = HashMap::new();
while let Ok(mut ev) = events.recv() {
let key = ev
.body_json
.as_ref()
.and_then(|b| b.get("correlate_key"))
.and_then(serde_json::Value::as_str)
.map(str::to_string);
match ev.kind {
EventKind::ToolCall => {
if let Ok(w) = writer.lock() {
if let Ok(id) = w.append_event(ev) {
if let Some(k) = key {
calls.insert(k, id);
}
}
}
continue;
}
EventKind::ToolResult => {
if let Some(k) = &key {
ev.correlates = calls.get(k).copied();
}
}
_ => {}
}
if let Ok(w) = writer.lock() {
let _ = w.append_event(ev);
}
}
}
fn flush_terminal_chunk(
writer: &Arc<Mutex<EventWriter>>,
blobs: &BlobStore,
session_id: &str,
start: Instant,
bytes: &[u8],
) {
let ts_ms = i64::try_from(start.elapsed().as_millis()).unwrap_or(i64::MAX);
let (body, blob_hash, blob_size) = match std::str::from_utf8(bytes) {
Ok(text) => (
serde_json::json!({ "text": text, "encoding": "utf8" }),
None,
None,
),
Err(_) => match blobs.put(bytes) {
Ok(outcome) => (
serde_json::json!({ "encoding": "blob", "size": outcome.size }),
Some(outcome.hash),
Some(outcome.size),
),
Err(_) => (
serde_json::json!({ "encoding": "base64", "bytes": base64_encode(bytes) }),
None,
None,
),
},
};
let event = Event {
session_id: session_id.to_string(),
ts_ms,
kind: EventKind::TerminalOutput,
step: None, summary: truncate_summary(&format!("terminal output {} bytes", bytes.len())),
body_json: Some(body),
blob_hash,
blob_size,
correlates: None,
};
if let Ok(w) = writer.lock() {
let _ = w.append_event(event);
}
}
fn flush_input_chunk(
writer: &Arc<Mutex<EventWriter>>,
session_id: &str,
start: Instant,
bytes: &[u8],
) {
let ts_ms = i64::try_from(start.elapsed().as_millis()).unwrap_or(i64::MAX);
let body = match std::str::from_utf8(bytes) {
Ok(text) => serde_json::json!({ "text": text, "encoding": "utf8", "direction": "input" }),
Err(_) => serde_json::json!({
"encoding": "base64",
"bytes": base64_encode(bytes),
"direction": "input",
}),
};
let event = Event {
session_id: session_id.to_string(),
ts_ms,
kind: EventKind::TerminalOutput,
step: None,
summary: truncate_summary(&format!("terminal input {} bytes", bytes.len())),
body_json: Some(body),
blob_hash: None,
blob_size: None,
correlates: None,
};
if let Ok(w) = writer.lock() {
let _ = w.append_event(event);
}
}
fn finalize(
store: &Store,
session_id: &str,
writer: &Arc<Mutex<EventWriter>>,
adapter_outcome: Option<&AdapterOutcome>,
exit_code: Option<i32>,
status: &str,
) -> crate::Result<(i64, i64, i64)> {
let status_enum = match status {
"ok" => hh_core::event::SessionStatus::Ok,
"error" => hh_core::event::SessionStatus::Error,
_ => hh_core::event::SessionStatus::Interrupted,
};
{
let mut w = writer
.lock()
.map_err(|e| crate::RecordError::Pty(format!("writer lock poisoned: {e}")))?;
w.flush().map_err(crate::RecordError::from)?;
w.close().map_err(crate::RecordError::from)?;
}
store.assign_steps(session_id)?;
if let Some(o) = adapter_outcome {
store.set_session_adapter_meta(
session_id,
o.model.as_deref(),
o.usage_json.as_ref(),
o.status,
)?;
}
let ended_at = now_unix_ms();
store.finalize_session(session_id, ended_at, exit_code, status_enum)?;
let (event_count, files_changed) = store.session_stats(session_id)?;
let steps = store.session_step_count(session_id)?;
Ok((event_count, steps, files_changed))
}
fn current_pty_size() -> PtySize {
match crossterm::terminal::size() {
Ok((cols, rows)) => PtySize {
rows,
cols,
pixel_width: 0,
pixel_height: 0,
},
Err(_) => PtySize::default(),
}
}
fn now_unix_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
}
fn hostname() -> Option<String> {
Command::new("hostname")
.output()
.ok()
.filter(|o| o.status.success())
.and_then(|o| String::from_utf8(o.stdout).ok())
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
}
fn truncate_summary(s: &str) -> String {
const LIMIT: usize = 120;
if s.chars().count() <= LIMIT {
return s.to_string();
}
let truncated: String = s.chars().take(LIMIT - 1).collect();
format!("{truncated}…")
}
fn base64_encode(bytes: &[u8]) -> String {
const TABLE: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
for chunk in bytes.chunks(3) {
let b0 = chunk[0];
let b1 = chunk.get(1).copied().unwrap_or(0);
let b2 = chunk.get(2).copied().unwrap_or(0);
let triple = (u32::from(b0) << 16) | (u32::from(b1) << 8) | u32::from(b2);
out.push(TABLE[((triple >> 18) & 0x3F) as usize] as char);
out.push(TABLE[((triple >> 12) & 0x3F) as usize] as char);
if chunk.len() > 1 {
out.push(TABLE[((triple >> 6) & 0x3F) as usize] as char);
} else {
out.push('=');
}
if chunk.len() > 2 {
out.push(TABLE[(triple & 0x3F) as usize] as char);
} else {
out.push('=');
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn base64_roundtrip_ascii() {
let s = base64_encode(b"hello");
assert_eq!(s, "aGVsbG8=");
}
#[test]
fn base64_handles_empty_and_padding() {
assert_eq!(base64_encode(b""), "");
assert_eq!(base64_encode(b"f"), "Zg==");
assert_eq!(base64_encode(b"fo"), "Zm8=");
assert_eq!(base64_encode(b"foo"), "Zm9v");
}
#[test]
fn truncate_summary_respects_limit() {
assert_eq!(truncate_summary("short"), "short");
let long: String = "x".repeat(200);
let t = truncate_summary(&long);
assert!(t.chars().count() <= 120);
assert!(t.ends_with('…'));
}
#[test]
fn now_unix_ms_is_plausible() {
let ms = now_unix_ms();
assert!(ms > 1_767_000_000_000);
assert!(ms < 4_000_000_000_000);
}
}