Pass sessions directly to session tasks (#36354)

## What changed

- Replace `SessionTaskContext` with `Arc<Session>` in the session task run and abort interfaces.
- Read extension data from `TurnContext` during regular turns instead of storing and forwarding a duplicate reference through the running task.

GitOrigin-RevId: 04bc0d324e5ba781f35ac9c61e811888208b74ae
This commit is contained in:
jif authored and copyberry committed 2026-07-31 16:05:52 +00:00
1 parent 2c005abb07
commit 35eab50501
8 files changed
+36 -106

No files matched your search

+10 -62
View File
@@ -8,7 +8,6 @@ use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use codex_extension_api::ExtensionData;
use futures::future::BoxFuture;
use tokio::select;
use tokio::sync::Notify;
@@ -36,8 +35,6 @@ use crate::state::RunningTask;
use crate::state::TaskKind;
use codex_analytics::TurnProfileFact;
use codex_analytics::TurnTokenUsageFact;
use codex_login::AuthManager;
use codex_models_manager::manager::SharedModelsManager;
use codex_otel::SessionTelemetry;
use codex_otel::TURN_E2E_DURATION_METRIC;
use codex_otel::TURN_MEMORY_METRIC;
@@ -176,38 +173,6 @@ fn bool_tag(value: bool) -> &'static str {
if value { "true" } else { "false" }
}
/// Thin wrapper that exposes the parts of [`Session`] task runners need.
#[derive(Clone)]
pub(crate) struct SessionTaskContext {
session: Arc<Session>,
turn_extension_data: Arc<ExtensionData>,
}
impl SessionTaskContext {
pub(crate) fn new(session: Arc<Session>, turn_extension_data: Arc<ExtensionData>) -> Self {
Self {
session,
turn_extension_data,
}
}
pub(crate) fn clone_session(&self) -> Arc<Session> {
Arc::clone(&self.session)
}
pub(crate) fn turn_extension_data(&self) -> Arc<ExtensionData> {
Arc::clone(&self.turn_extension_data)
}
pub(crate) fn auth_manager(&self) -> Arc<AuthManager> {
Arc::clone(&self.session.services.auth_manager)
}
pub(crate) fn models_manager(&self) -> SharedModelsManager {
Arc::clone(&self.session.services.models_manager)
}
}
/// Async task that drives a [`Session`] turn.
///
/// Implementations encapsulate a specific Codex workflow (regular chat,
@@ -236,7 +201,7 @@ pub(crate) trait SessionTask: Send + Sync + 'static {
/// lifecycle instead.
fn run(
self: Arc<Self>,
session: Arc<SessionTaskContext>,
session: Arc<Session>,
ctx: Arc<TurnContext>,
input: Vec<TurnInput>,
cancellation_token: CancellationToken,
@@ -249,7 +214,7 @@ pub(crate) trait SessionTask: Send + Sync + 'static {
/// [`Session::abort_all_tasks`] cancels the task.
fn abort(
&self,
session: Arc<SessionTaskContext>,
session: Arc<Session>,
ctx: Arc<TurnContext>,
) -> impl std::future::Future<Output = ()> + Send {
async move {
@@ -265,17 +230,13 @@ pub(crate) trait AnySessionTask: Send + Sync + 'static {
fn run(
self: Arc<Self>,
session: Arc<SessionTaskContext>,
session: Arc<Session>,
ctx: Arc<TurnContext>,
input: Vec<TurnInput>,
cancellation_token: CancellationToken,
) -> BoxFuture<'static, SessionTaskResult>;
fn abort<'a>(
&'a self,
session: Arc<SessionTaskContext>,
ctx: Arc<TurnContext>,
) -> BoxFuture<'a, ()>;
fn abort<'a>(&'a self, session: Arc<Session>, ctx: Arc<TurnContext>) -> BoxFuture<'a, ()>;
}
impl<T> AnySessionTask for T
@@ -292,7 +253,7 @@ where
fn run(
self: Arc<Self>,
session: Arc<SessionTaskContext>,
session: Arc<Session>,
ctx: Arc<TurnContext>,
input: Vec<TurnInput>,
cancellation_token: CancellationToken,
@@ -306,11 +267,7 @@ where
))
}
fn abort<'a>(
&'a self,
session: Arc<SessionTaskContext>,
ctx: Arc<TurnContext>,
) -> BoxFuture<'a, ()> {
fn abort<'a>(&'a self, session: Arc<Session>, ctx: Arc<TurnContext>) -> BoxFuture<'a, ()> {
Box::pin(SessionTask::abort(self, session, ctx))
}
}
@@ -377,7 +334,6 @@ impl Session {
self.emit_turn_start_lifecycle(turn_context.as_ref(), &token_usage_at_turn_start)
.await;
let turn_extension_data = Arc::clone(&turn_context.extension_data);
let mut active = self.active_turn.lock().await;
let turn = active.get_or_insert_with(ActiveTurn::default);
debug_assert!(turn.task.is_none());
@@ -386,10 +342,7 @@ impl Session {
&turn_context.session_source,
);
let done_clone = Arc::clone(&done);
let session_ctx = Arc::new(SessionTaskContext::new(
Arc::clone(self),
Arc::clone(&turn_extension_data),
));
let session = Arc::clone(self);
let ctx = Arc::clone(&turn_context);
let task_for_run = Arc::clone(&task);
let task_input = input;
@@ -417,14 +370,14 @@ impl Session {
let ctx_for_finish = Arc::clone(&ctx);
let task_result = task_for_run
.run(
Arc::clone(&session_ctx),
Arc::clone(&session),
ctx,
task_input,
task_cancellation_token.child_token(),
)
.instrument(trace_span!("session_task.run"))
.await;
let sess = session_ctx.clone_session();
let sess = Arc::clone(&session);
if let Err(err) = sess.flush_rollout().await {
warn!("failed to flush rollout before completing turn: {err}");
sess.send_event(
@@ -457,7 +410,6 @@ impl Session {
task,
cancellation_token,
turn_context: Arc::clone(&turn_context),
turn_extension_data,
_agent_execution_guard: agent_execution_guard,
_timer: timer,
};
@@ -917,12 +869,8 @@ impl Session {
task.handle.abort();
let session_ctx = Arc::new(SessionTaskContext::new(
Arc::clone(self),
Arc::clone(&task.turn_extension_data),
));
session_task
.abort(session_ctx, Arc::clone(&task.turn_context))
.abort(Arc::clone(self), Arc::clone(&task.turn_context))
.await;
if reason == TurnAbortReason::Interrupted