[WIP] Name conversations

This commit is contained in:
Javier Soto
2026-01-05 11:08:53 -08:00
parent 720fa67816
commit 1364a81040
16 changed files with 223 additions and 4 deletions
+42
View File
@@ -110,6 +110,7 @@ use crate::protocol::ReasoningRawContentDeltaEvent;
use crate::protocol::ReviewDecision;
use crate::protocol::SandboxPolicy;
use crate::protocol::SessionConfiguredEvent;
use crate::protocol::SessionRenamedEvent;
use crate::protocol::SkillErrorInfo;
use crate::protocol::SkillMetadata as ProtocolSkillMetadata;
use crate::protocol::StreamErrorEvent;
@@ -1319,6 +1320,17 @@ impl Session {
}
}
pub(crate) async fn set_session_name(&self, name: String) -> std::io::Result<()> {
let recorder = {
let guard = self.services.rollout.lock().await;
guard.clone()
};
let Some(rec) = recorder else {
return Err(std::io::Error::other("rollout recorder unavailable"));
};
rec.set_session_name(name).await
}
pub(crate) async fn clone_history(&self) -> ContextManager {
let state = self.state.lock().await;
state.clone_history()
@@ -1658,6 +1670,9 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
Op::Compact => {
handlers::compact(&sess, sub.id.clone()).await;
}
Op::SetSessionName { name } => {
handlers::set_session_name(&sess, sub.id.clone(), name).await;
}
Op::RunUserShellCommand { command } => {
handlers::run_user_shell_command(
&sess,
@@ -1714,6 +1729,7 @@ mod handlers {
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::SessionRenamedEvent;
use codex_protocol::protocol::SkillsListEntry;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::WarningEvent;
@@ -2034,6 +2050,32 @@ mod handlers {
.await;
}
pub async fn set_session_name(sess: &Arc<Session>, sub_id: String, name: String) {
let trimmed = name.trim();
if trimmed.is_empty() {
return;
}
if let Err(err) = sess.set_session_name(trimmed.to_string()).await {
let message = format!("Failed to update session name: {err}");
warn!("{message}");
let event = Event {
id: sub_id,
msg: EventMsg::Warning(WarningEvent { message }),
};
sess.send_event_raw(event).await;
return;
}
let event = Event {
id: sub_id,
msg: EventMsg::SessionRenamed(SessionRenamedEvent {
name: trimmed.to_string(),
}),
};
sess.send_event_raw(event).await;
}
pub async fn shutdown(sess: &Arc<Session>, sub_id: String) -> bool {
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
sess.services
+1
View File
@@ -55,6 +55,7 @@ pub(crate) fn should_persist_event_msg(ev: &EventMsg) -> bool {
| EventMsg::AgentReasoningDelta(_)
| EventMsg::AgentReasoningRawContentDelta(_)
| EventMsg::AgentReasoningSectionBreak(_)
| EventMsg::SessionRenamed(_)
| EventMsg::RawResponseItem(_)
| EventMsg::SessionConfigured(_)
| EventMsg::McpToolCallBegin(_)
+61 -1
View File
@@ -11,6 +11,7 @@ use serde_json::Value;
use time::OffsetDateTime;
use time::format_description::FormatItem;
use time::macros::format_description;
use tokio::io::AsyncSeekExt;
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc::Sender;
use tokio::sync::mpsc::{self};
@@ -67,6 +68,11 @@ enum RolloutCmd {
Flush {
ack: oneshot::Sender<()>,
},
/// Update the session meta name stored at the head of the rollout file.
UpdateSessionName {
name: String,
ack: oneshot::Sender<std::io::Result<()>>,
},
Shutdown {
ack: oneshot::Sender<()>,
},
@@ -146,6 +152,7 @@ impl RolloutRecorder {
originator: originator().value.clone(),
cli_version: env!("CARGO_PKG_VERSION").to_string(),
instructions,
name: None,
source,
model_provider: Some(config.model_provider_id.clone()),
}),
@@ -172,7 +179,7 @@ impl RolloutRecorder {
// Spawn a Tokio task that owns the file handle and performs async
// writes. Using `tokio::fs::File` keeps everything on the async I/O
// driver instead of blocking the runtime.
tokio::task::spawn(rollout_writer(file, rx, meta, cwd));
tokio::task::spawn(rollout_writer(file, rx, meta, cwd, rollout_path.clone()));
Ok(Self { tx, rollout_path })
}
@@ -207,6 +214,16 @@ impl RolloutRecorder {
.map_err(|e| IoError::other(format!("failed waiting for rollout flush: {e}")))
}
pub async fn set_session_name(&self, name: String) -> std::io::Result<()> {
let (tx, rx) = oneshot::channel();
self.tx
.send(RolloutCmd::UpdateSessionName { name, ack: tx })
.await
.map_err(|e| IoError::other(format!("failed to queue session name update: {e}")))?;
rx.await
.map_err(|e| IoError::other(format!("failed waiting for session name update: {e}")))?
}
pub async fn get_rollout_history(path: &Path) -> std::io::Result<InitialHistory> {
info!("Resuming rollout from {path:?}");
let text = tokio::fs::read_to_string(path).await?;
@@ -351,6 +368,7 @@ async fn rollout_writer(
mut rx: mpsc::Receiver<RolloutCmd>,
mut meta: Option<SessionMeta>,
cwd: std::path::PathBuf,
rollout_path: PathBuf,
) -> std::io::Result<()> {
let mut writer = JsonlWriter { file };
@@ -389,6 +407,10 @@ async fn rollout_writer(
RolloutCmd::Shutdown { ack } => {
let _ = ack.send(());
}
RolloutCmd::UpdateSessionName { name, ack } => {
let result = update_session_name(&mut writer, &rollout_path, name).await;
let _ = ack.send(result);
}
}
}
@@ -422,3 +444,41 @@ impl JsonlWriter {
Ok(())
}
}
async fn update_session_name(
writer: &mut JsonlWriter,
rollout_path: &Path,
name: String,
) -> std::io::Result<()> {
writer.file.flush().await?;
let contents = tokio::fs::read(rollout_path).await?;
let newline_idx = contents
.iter()
.position(|&b| b == b'\n')
.ok_or_else(|| IoError::other("rollout is missing a SessionMeta line"))?;
let first_line = std::str::from_utf8(&contents[..newline_idx])
.map_err(|e| IoError::other(format!("invalid utf8 in session meta line: {e}")))?;
let mut rollout_line: RolloutLine = serde_json::from_str(first_line)
.map_err(|e| IoError::other(format!("invalid session meta line: {e}")))?;
let RolloutItem::SessionMeta(mut session_meta_line) = rollout_line.item else {
return Err(IoError::other("first rollout line is not session metadata"));
};
let trimmed = name.trim();
session_meta_line.meta.name = if trimmed.is_empty() {
None
} else {
Some(trimmed.to_string())
};
rollout_line.item = RolloutItem::SessionMeta(session_meta_line);
let mut updated = serde_json::to_vec(&rollout_line)?;
updated.push(b'\n');
updated.extend_from_slice(&contents[newline_idx + 1..]);
writer.file.set_len(0).await?;
writer.file.seek(std::io::SeekFrom::Start(0)).await?;
writer.file.write_all(&updated).await?;
writer.file.flush().await?;
Ok(())
}
+1
View File
@@ -587,6 +587,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
id: conversation_id,
timestamp: ts.to_string(),
instructions: None,
name: None,
cwd: ".".into(),
originator: "test_originator".into(),
cli_version: "test_version".into(),