Expose thread APIs for message board access (#59154)

## What changed

- Add `CodexThread` methods for reading the registration source, resolving agent references, and reading the configured clock.
- Replace the board-specific thread lookup with public `ThreadManager::get_thread_with_extension`, which requires matching typed extension data and never loads a missing runtime.
- Update the local message board host to use these APIs and thread startup metadata instead of accessing session internals directly.

GitOrigin-RevId: 143b798cc710e079e20be59a84d05738a3bc6a0b
This commit is contained in:
jif
2026-11-03 13:17:38 +00:00
committed by copyberry
parent f05aa41da6
commit 4a971492e7
3 changed files with 56 additions and 49 deletions
+12 -34
View File
@@ -116,10 +116,14 @@ impl LocalBoardHost {
.upgrade()
.ok_or_else(|| CodexErr::ThreadNotFound(caller))?;
if let Some(connection) = &self.connection {
return manager.get_thread_for_board(caller, connection).await;
return manager
.get_thread_with_extension::<LocalBoardConnection>(caller, |other| {
connection.same_board(other)
})
.await;
}
let actor = manager.get_thread(caller).await?;
if actor.session.session_id() != self.tree
if actor.startup_metadata().session_id != self.tree
|| actor
.thread_extension_data()
.get::<LocalBoardConnection>()
@@ -133,26 +137,16 @@ impl LocalBoardHost {
}
async fn registered_path(&self, actor: &CodexThread) -> Result<AgentPath> {
let caller = actor.session.thread_id;
let caller = actor.startup_metadata().thread_id;
if let Some(connection) = &self.connection {
return connection.caller_path(caller);
}
let path = actor
.session_source
.session_source()
.get_agent_path()
.or_else(|| (caller == ThreadId::from(self.tree)).then(AgentPath::root))
.ok_or_else(|| CodexErr::InvalidRequest("agent has no tree path".into()))?;
let resolved = actor
.session
.services
.agent_control
.resolve(
caller,
actor.session_source.parent_thread_id(),
&actor.session_source,
path.as_str(),
)
.await?;
let resolved = actor.resolve_agent_reference(path.as_str()).await?;
if resolved != caller {
return Err(CodexErr::InvalidRequest(
"agent does not own its tree path".into(),
@@ -183,20 +177,10 @@ impl MessageBoardHost for LocalBoardHost {
return self
.actor(caller)
.await
.map(|actor| actor.session.thread_id);
.map(|actor| actor.startup_metadata().thread_id);
}
let actor = self.actor(self.caller).await?;
actor
.session
.services
.agent_control
.resolve(
self.caller,
actor.session_source.parent_thread_id(),
&actor.session_source,
path.as_str(),
)
.await
actor.resolve_agent_reference(path.as_str()).await
})
}
@@ -204,13 +188,7 @@ impl MessageBoardHost for LocalBoardHost {
Box::pin(async move {
let actor = self.actor(caller).await?;
self.registered_path(&actor).await?;
actor
.session
.services
.time_provider
.current_time(caller)
.await
.map_err(|error| CodexErr::Io(std::io::Error::other(error)))
actor.current_time().await
})
}
+30
View File
@@ -245,6 +245,36 @@ impl CodexThread {
self.submit_with_trace(op, /*trace*/ None).await
}
/// Returns the source used for registration and agent-reference resolution.
/// Internal runtimes can have a different source in their captured session settings.
pub fn session_source(&self) -> &SessionSource {
&self.session_source
}
/// Resolves an agent reference through this thread's controller without loading it.
pub async fn resolve_agent_reference(&self, target: &str) -> CodexResult<ThreadId> {
self.session
.services
.agent_control
.resolve(
self.session.thread_id,
self.session_source.parent_thread_id(),
&self.session_source,
target,
)
.await
}
/// Reads this thread's configured clock, including an external or simulated clock.
pub async fn current_time(&self) -> CodexResult<DateTime<Utc>> {
self.session
.services
.time_provider
.current_time(self.session.thread_id)
.await
.map_err(|error| CodexErr::Io(std::io::Error::other(error)))
}
/// Returns the session telemetry handle for thread-scoped production instrumentation.
pub fn session_telemetry(&self) -> SessionTelemetry {
self.session.services.session_telemetry.clone()
+14 -15
View File
@@ -1025,25 +1025,24 @@ impl ThreadManager {
}
}
/// Board hosts may reach internal runtimes only through a matching explicit connection.
pub(crate) async fn get_thread_for_board(
/// Finds a registered runtime whose host-owned extension data matches the supplied scope.
/// Unlike `get_thread`, this can return internal runtimes, but only after validating their
/// explicit extension attachment. It never loads a missing runtime.
pub async fn get_thread_with_extension<T: Send + Sync + 'static>(
&self,
thread_id: ThreadId,
connection: &codex_message_board::LocalBoardConnection,
matches: impl FnOnce(&T) -> bool,
) -> CodexResult<Arc<CodexThread>> {
self.state
.threads
.read()
.await
let threads = self.state.threads.read().await;
let thread = threads
.get(&thread_id)
.filter(|thread| {
thread
.thread_extension_data()
.get::<codex_message_board::LocalBoardConnection>()
.is_some_and(|other| connection.same_board(&other))
})
.cloned()
.ok_or(CodexErr::ThreadNotFound(thread_id))
.ok_or(CodexErr::ThreadNotFound(thread_id))?;
let extension = thread.thread_extension_data().get::<T>();
if extension.as_deref().is_some_and(matches) {
Ok(Arc::clone(thread))
} else {
Err(CodexErr::ThreadNotFound(thread_id))
}
}
/// Updates metadata for loaded and cold threads through one entrypoint.