Merge 1a6aba2f88 into sapling-pr-archive-bolinfest

This commit is contained in:
Michael Bolin
2026-05-28 09:19:35 -07:00
committed by GitHub
47 changed files with 1584 additions and 266 deletions
+11 -2
View File
@@ -150,7 +150,9 @@ for arg in "\$@"; do
args+=("\${arg}")
done
exec "${zig_bin}" cc -target "${zig_target}" "\${args[@]}"
# Zig enables UBSan for debug C builds by default. Rust links these objects
# without Zig's sanitizer runtime, so keep native dependencies uninstrumented.
exec "${zig_bin}" cc -target "${zig_target}" "\${args[@]}" -fno-sanitize=undefined
EOF
cat >"${cxx}" <<EOF
#!/usr/bin/env bash
@@ -207,7 +209,9 @@ for arg in "\$@"; do
args+=("\${arg}")
done
exec "${zig_bin}" c++ -target "${zig_target}" "\${args[@]}"
# Zig enables UBSan for debug C++ builds by default. Rust links these objects
# without Zig's sanitizer runtime, so keep native dependencies uninstrumented.
exec "${zig_bin}" c++ -target "${zig_target}" "\${args[@]}" -fno-sanitize=undefined
EOF
chmod +x "${cc}" "${cxx}"
@@ -270,6 +274,11 @@ echo "PKG_CONFIG_PATH=${pkg_config_path}" >> "$GITHUB_ENV"
pkg_config_path_var="PKG_CONFIG_PATH_${TARGET}"
pkg_config_path_var="${pkg_config_path_var//-/_}"
echo "${pkg_config_path_var}=${libcap_pkgconfig_dir}" >> "$GITHUB_ENV"
pkg_config_libdir_var="PKG_CONFIG_LIBDIR_${TARGET}"
pkg_config_libdir_var="${pkg_config_libdir_var//-/_}"
# Do not let musl cross-builds resolve native libraries from the host glibc
# pkg-config directories. libcap is the only target package provided here.
echo "${pkg_config_libdir_var}=${libcap_pkgconfig_dir}" >> "$GITHUB_ENV"
if [[ -n "${sysroot}" && "${sysroot}" != "/" ]]; then
echo "PKG_CONFIG_SYSROOT_DIR=${sysroot}" >> "$GITHUB_ENV"
+8 -65
View File
@@ -260,11 +260,7 @@ jobs:
set -euo pipefail
if command -v apt-get >/dev/null 2>&1; then
sudo apt-get update -y
packages=(pkg-config libcap-dev)
if [[ "${{ matrix.target }}" == 'x86_64-unknown-linux-musl' || "${{ matrix.target }}" == 'aarch64-unknown-linux-musl' ]]; then
packages+=(libubsan1)
fi
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends "${packages[@]}"
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends pkg-config libcap-dev
fi
- uses: dtolnay/rust-toolchain@e081816240890017053eacbb1bdf337761dc5582 # 1.95.0
with:
@@ -348,14 +344,6 @@ jobs:
sccache-${{ matrix.runner }}-${{ matrix.target }}-${{ matrix.profile }}-${{ steps.lockhash.outputs.hash }}-
sccache-${{ matrix.runner }}-${{ matrix.target }}-${{ matrix.profile }}-
- if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}}
name: Disable sccache wrapper (musl)
shell: bash
run: |
set -euo pipefail
echo "RUSTC_WRAPPER=" >> "$GITHUB_ENV"
echo "RUSTC_WORKSPACE_WRAPPER=" >> "$GITHUB_ENV"
- if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}}
name: Prepare APT cache directories (musl)
shell: bash
@@ -389,58 +377,6 @@ jobs:
shell: bash
run: bash "${GITHUB_WORKSPACE}/.github/scripts/install-musl-build-tools.sh"
- if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}}
name: Configure rustc UBSan wrapper (musl host)
shell: bash
run: |
set -euo pipefail
ubsan=""
if command -v ldconfig >/dev/null 2>&1; then
ubsan="$(ldconfig -p | grep -m1 'libubsan\.so\.1' | sed -E 's/.*=> (.*)$/\1/')"
fi
wrapper_root="${RUNNER_TEMP:-/tmp}"
wrapper="${wrapper_root}/rustc-ubsan-wrapper"
cat > "${wrapper}" <<EOF
#!/usr/bin/env bash
set -euo pipefail
if [[ -n "${ubsan}" ]]; then
export LD_PRELOAD="${ubsan}\${LD_PRELOAD:+:\${LD_PRELOAD}}"
fi
exec "\$1" "\${@:2}"
EOF
chmod +x "${wrapper}"
echo "RUSTC_WRAPPER=${wrapper}" >> "$GITHUB_ENV"
echo "RUSTC_WORKSPACE_WRAPPER=" >> "$GITHUB_ENV"
- if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}}
name: Clear sanitizer flags (musl)
shell: bash
run: |
set -euo pipefail
# Clear global Rust flags so host/proc-macro builds don't pull in UBSan.
echo "RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_ENCODED_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "RUSTDOCFLAGS=" >> "$GITHUB_ENV"
# Override any runner-level Cargo config rustflags as well.
echo "CARGO_BUILD_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_MUSL_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_RUSTFLAGS=" >> "$GITHUB_ENV"
sanitize_flags() {
local input="$1"
input="${input//-fsanitize=undefined/}"
input="${input//-fno-sanitize-recover=undefined/}"
input="${input//-fno-sanitize-trap=undefined/}"
echo "$input"
}
cflags="$(sanitize_flags "${CFLAGS-}")"
cxxflags="$(sanitize_flags "${CXXFLAGS-}")"
echo "CFLAGS=${cflags}" >> "$GITHUB_ENV"
echo "CXXFLAGS=${cxxflags}" >> "$GITHUB_ENV"
- if: ${{ !contains(matrix.target, 'windows') }}
name: Configure rusty_v8 artifact overrides and verify checksums
uses: ./.github/actions/setup-rusty-v8
@@ -466,6 +402,13 @@ jobs:
- name: cargo clippy
run: cargo clippy --target ${{ matrix.target }} --tests --profile ${{ matrix.profile }} --timings -- -D warnings
- uses: taiki-e/install-action@44c6d64aa62cd779e873306675c7a58e86d6d532 # v2.62.49
with:
tool: just
- name: End-to-end benchmark smoke test
run: just bench-e2e-smoke
- name: Upload Cargo timings (clippy)
if: always()
uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
+1 -57
View File
@@ -245,15 +245,6 @@ jobs:
set -euo pipefail
sudo apt-get update -y
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends pkg-config libcap-dev
- name: Install UBSan runtime (musl)
if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl' }}
shell: bash
run: |
set -euo pipefail
if command -v apt-get >/dev/null 2>&1; then
sudo apt-get update -y
sudo DEBIAN_FRONTEND=noninteractive apt-get install -y libubsan1
fi
- uses: dtolnay/rust-toolchain@e081816240890017053eacbb1bdf337761dc5582 # 1.95.0
with:
targets: ${{ matrix.target }}
@@ -283,30 +274,7 @@ jobs:
run: bash "${GITHUB_WORKSPACE}/.github/scripts/install-musl-build-tools.sh"
- if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}}
name: Configure rustc UBSan wrapper (musl host)
shell: bash
run: |
set -euo pipefail
ubsan=""
if command -v ldconfig >/dev/null 2>&1; then
ubsan="$(ldconfig -p | grep -m1 'libubsan\.so\.1' | sed -E 's/.*=> (.*)$/\1/')"
fi
wrapper_root="${RUNNER_TEMP:-/tmp}"
wrapper="${wrapper_root}/rustc-ubsan-wrapper"
cat > "${wrapper}" <<EOF
#!/usr/bin/env bash
set -euo pipefail
if [[ -n "${ubsan}" ]]; then
export LD_PRELOAD="${ubsan}\${LD_PRELOAD:+:\${LD_PRELOAD}}"
fi
exec "\$1" "\${@:2}"
EOF
chmod +x "${wrapper}"
echo "RUSTC_WRAPPER=${wrapper}" >> "$GITHUB_ENV"
echo "RUSTC_WORKSPACE_WRAPPER=" >> "$GITHUB_ENV"
- if: ${{ matrix.target == 'x86_64-unknown-linux-musl' || matrix.target == 'aarch64-unknown-linux-musl'}}
name: Clear sanitizer flags (musl)
name: Disable aws-lc jitter entropy (musl)
shell: bash
run: |
set -euo pipefail
@@ -316,30 +284,6 @@ jobs:
target_no_jitter="${target_no_jitter//-/_}"
echo "${target_no_jitter}=1" >> "$GITHUB_ENV"
# Clear global Rust flags so host/proc-macro builds don't pull in UBSan.
echo "RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_ENCODED_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "RUSTDOCFLAGS=" >> "$GITHUB_ENV"
# Override any runner-level Cargo config rustflags as well.
echo "CARGO_BUILD_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_TARGET_X86_64_UNKNOWN_LINUX_MUSL_RUSTFLAGS=" >> "$GITHUB_ENV"
echo "CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_RUSTFLAGS=" >> "$GITHUB_ENV"
sanitize_flags() {
local input="$1"
input="${input//-fsanitize=undefined/}"
input="${input//-fno-sanitize-recover=undefined/}"
input="${input//-fno-sanitize-trap=undefined/}"
echo "$input"
}
cflags="$(sanitize_flags "${CFLAGS-}")"
cxxflags="$(sanitize_flags "${CXXFLAGS-}")"
echo "CFLAGS=${cflags}" >> "$GITHUB_ENV"
echo "CXXFLAGS=${cxxflags}" >> "$GITHUB_ENV"
- name: Configure rusty_v8 artifact overrides and verify checksums
uses: ./.github/actions/setup-rusty-v8
with:
+11
View File
@@ -2058,6 +2058,17 @@ dependencies = [
"uuid",
]
[[package]]
name = "codex-app-server-start-bench"
version = "0.0.0"
dependencies = [
"anyhow",
"codex-app-server-protocol",
"divan",
"serde_json",
"tempfile",
]
[[package]]
name = "codex-app-server-test-client"
version = "0.0.0"
+1
View File
@@ -14,6 +14,7 @@ members = [
"app-server-client",
"app-server-protocol",
"app-server-test-client",
"benchmarks/app-server-start",
"debug-client",
"apply-patch",
"arg0",
@@ -263,6 +263,7 @@ fn sample_thread_resume_response_with_source(
sandbox: AppServerSandboxPolicy::DangerFullAccess,
active_permission_profile: None,
reasoning_effort: None,
initial_turns_page: None,
})
}
+1
View File
@@ -172,6 +172,7 @@ fn sample_thread_resume_response() -> ClientResponsePayload {
sandbox: AppServerSandboxPolicy::DangerFullAccess,
active_permission_profile: None,
reasoning_effort: None,
initial_turns_page: None,
})
}
@@ -3543,6 +3543,42 @@
}
]
},
"ThreadResumeInitialTurnsPageParams": {
"properties": {
"itemsView": {
"anyOf": [
{
"$ref": "#/definitions/TurnItemsView"
},
{
"type": "null"
}
],
"description": "How much item detail to include for each returned turn; defaults to summary."
},
"limit": {
"description": "Optional turn page size.",
"format": "uint32",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional turn pagination direction; defaults to descending."
}
},
"type": "object"
},
"ThreadResumeParams": {
"description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.",
"properties": {
@@ -17150,6 +17150,42 @@
"title": "ThreadRealtimeTranscriptDoneNotification",
"type": "object"
},
"ThreadResumeInitialTurnsPageParams": {
"properties": {
"itemsView": {
"anyOf": [
{
"$ref": "#/definitions/v2/TurnItemsView"
},
{
"type": "null"
}
],
"description": "How much item detail to include for each returned turn; defaults to summary."
},
"limit": {
"description": "Optional turn page size.",
"format": "uint32",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/v2/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional turn pagination direction; defaults to descending."
}
},
"type": "object"
},
"ThreadResumeParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.",
@@ -18488,6 +18524,32 @@
"title": "TurnSteerResponse",
"type": "object"
},
"TurnsPage": {
"properties": {
"backwardsCursor": {
"type": [
"string",
"null"
]
},
"data": {
"items": {
"$ref": "#/definitions/v2/Turn"
},
"type": "array"
},
"nextCursor": {
"type": [
"string",
"null"
]
}
},
"required": [
"data"
],
"type": "object"
},
"UserInput": {
"oneOf": [
{
@@ -14974,6 +14974,42 @@
"title": "ThreadRealtimeTranscriptDoneNotification",
"type": "object"
},
"ThreadResumeInitialTurnsPageParams": {
"properties": {
"itemsView": {
"anyOf": [
{
"$ref": "#/definitions/TurnItemsView"
},
{
"type": "null"
}
],
"description": "How much item detail to include for each returned turn; defaults to summary."
},
"limit": {
"description": "Optional turn page size.",
"format": "uint32",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional turn pagination direction; defaults to descending."
}
},
"type": "object"
},
"ThreadResumeParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.",
@@ -16312,6 +16348,32 @@
"title": "TurnSteerResponse",
"type": "object"
},
"TurnsPage": {
"properties": {
"backwardsCursor": {
"type": [
"string",
"null"
]
},
"data": {
"items": {
"$ref": "#/definitions/Turn"
},
"type": "array"
},
"nextCursor": {
"type": [
"string",
"null"
]
}
},
"required": [
"data"
],
"type": "object"
},
"UserInput": {
"oneOf": [
{
@@ -983,6 +983,74 @@
"danger-full-access"
],
"type": "string"
},
"SortDirection": {
"enum": [
"asc",
"desc"
],
"type": "string"
},
"ThreadResumeInitialTurnsPageParams": {
"properties": {
"itemsView": {
"anyOf": [
{
"$ref": "#/definitions/TurnItemsView"
},
{
"type": "null"
}
],
"description": "How much item detail to include for each returned turn; defaults to summary."
},
"limit": {
"description": "Optional turn page size.",
"format": "uint32",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional turn pagination direction; defaults to descending."
}
},
"type": "object"
},
"TurnItemsView": {
"oneOf": [
{
"description": "`items` was not loaded for this turn. The field is intentionally empty.",
"enum": [
"notLoaded"
],
"type": "string"
},
{
"description": "`items` contains only a display summary for this turn.",
"enum": [
"summary"
],
"type": "string"
},
{
"description": "`items` contains every ThreadItem available from persisted app-server history for this turn.",
"enum": [
"full"
],
"type": "string"
}
]
}
},
"description": "There are three ways to resume a thread: 1. By thread_id: load the thread from disk by thread_id and resume it. 2. By history: instantiate the thread from memory and resume it. 3. By path: load the thread from disk by path and resume it.\n\nFor non-running threads, the precedence is: history > non-empty path > thread_id. If using history or a non-empty path for a non-running thread, the thread_id param will be ignored.\n\nIf thread_id identifies a running thread, app-server rejoins that thread and treats a non-empty path as a consistency check against the active rollout path. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.",
@@ -1995,6 +1995,32 @@
],
"type": "string"
},
"TurnsPage": {
"properties": {
"backwardsCursor": {
"type": [
"string",
"null"
]
},
"data": {
"items": {
"$ref": "#/definitions/Turn"
},
"type": "array"
},
"nextCursor": {
"type": [
"string",
"null"
]
}
},
"required": [
"data"
],
"type": "object"
},
"UserInput": {
"oneOf": [
{
@@ -0,0 +1,19 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { SortDirection } from "./SortDirection";
import type { TurnItemsView } from "./TurnItemsView";
export type ThreadResumeInitialTurnsPageParams = {
/**
* Optional turn page size.
*/
limit?: number | null,
/**
* Optional turn pagination direction; defaults to descending.
*/
sortDirection?: SortDirection | null,
/**
* How much item detail to include for each returned turn; defaults to summary.
*/
itemsView?: TurnItemsView | null, };
@@ -0,0 +1,6 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { Turn } from "./Turn";
export type TurnsPage = { data: Array<Turn>, nextCursor: string | null, backwardsCursor: string | null, };
@@ -398,6 +398,7 @@ export type { ThreadRealtimeStartTransport } from "./ThreadRealtimeStartTranspor
export type { ThreadRealtimeStartedNotification } from "./ThreadRealtimeStartedNotification";
export type { ThreadRealtimeTranscriptDeltaNotification } from "./ThreadRealtimeTranscriptDeltaNotification";
export type { ThreadRealtimeTranscriptDoneNotification } from "./ThreadRealtimeTranscriptDoneNotification";
export type { ThreadResumeInitialTurnsPageParams } from "./ThreadResumeInitialTurnsPageParams";
export type { ThreadResumeParams } from "./ThreadResumeParams";
export type { ThreadResumeResponse } from "./ThreadResumeResponse";
export type { ThreadRollbackParams } from "./ThreadRollbackParams";
@@ -450,6 +451,7 @@ export type { TurnStartedNotification } from "./TurnStartedNotification";
export type { TurnStatus } from "./TurnStatus";
export type { TurnSteerParams } from "./TurnSteerParams";
export type { TurnSteerResponse } from "./TurnSteerResponse";
export type { TurnsPage } from "./TurnsPage";
export type { UserInput } from "./UserInput";
export type { WarningNotification } from "./WarningNotification";
export type { WebSearchAction } from "./WebSearchAction";
@@ -111,6 +111,85 @@ fn thread_turns_list_params_accepts_items_view() {
assert_eq!(params.items_view, Some(TurnItemsView::NotLoaded));
}
#[test]
fn thread_resume_params_accept_turns_page_bootstrap() {
let params = serde_json::from_value::<ThreadResumeParams>(json!({
"threadId": "thr_123",
"initialTurnsPage": {
"limit": 25,
"sortDirection": "asc",
"itemsView": "full",
},
}))
.expect("thread resume params should deserialize");
assert_eq!(params.thread_id, "thr_123");
assert_eq!(
params.initial_turns_page,
Some(ThreadResumeInitialTurnsPageParams {
limit: Some(25),
sort_direction: Some(SortDirection::Asc),
items_view: Some(TurnItemsView::Full),
})
);
}
#[test]
fn thread_resume_response_round_trips_initial_turns_page() {
let response = ThreadResumeResponse {
thread: Thread {
id: "thr_123".to_string(),
session_id: "thr_123".to_string(),
forked_from_id: None,
preview: String::new(),
ephemeral: false,
model_provider: "openai".to_string(),
created_at: 1,
updated_at: 1,
status: ThreadStatus::Idle,
path: None,
cwd: absolute_path("tmp"),
cli_version: "0.0.0".to_string(),
source: SessionSource::Exec,
thread_source: None,
agent_nickname: None,
agent_role: None,
git_info: None,
name: None,
turns: Vec::new(),
},
model: "gpt-5".to_string(),
model_provider: "openai".to_string(),
service_tier: None,
cwd: absolute_path("tmp"),
runtime_workspace_roots: Vec::new(),
instruction_sources: Vec::new(),
approval_policy: AskForApproval::OnFailure,
approvals_reviewer: ApprovalsReviewer::User,
sandbox: SandboxPolicy::DangerFullAccess,
active_permission_profile: None,
reasoning_effort: None,
initial_turns_page: Some(TurnsPage {
data: Vec::new(),
next_cursor: Some("cursor_next".to_string()),
backwards_cursor: Some("cursor_back".to_string()),
}),
};
let value = serde_json::to_value(&response).expect("serialize thread resume response");
assert_eq!(
value.get("initialTurnsPage"),
Some(&json!({
"data": [],
"nextCursor": "cursor_next",
"backwardsCursor": "cursor_back",
}))
);
let decoded = serde_json::from_value::<ThreadResumeResponse>(value)
.expect("deserialize thread resume response");
assert_eq!(decoded, response);
}
#[test]
fn thread_turns_items_list_round_trips() {
let params = ThreadTurnsItemsListParams {
@@ -3406,6 +3485,7 @@ fn thread_lifecycle_responses_default_missing_optional_fields() {
assert_eq!(fork.instruction_sources, Vec::<AbsolutePathBuf>::new());
assert_eq!(start.active_permission_profile, None);
assert_eq!(resume.active_permission_profile, None);
assert_eq!(resume.initial_turns_page, None);
assert_eq!(fork.active_permission_profile, None);
}
@@ -397,6 +397,11 @@ pub struct ThreadResumeParams {
#[experimental("thread/resume.excludeTurns")]
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub exclude_turns: bool,
/// When present, include a `thread/turns/list` page in the resume response
/// so clients can bootstrap recent turns without a second request.
#[experimental("thread/resume.initialTurnsPage")]
#[ts(optional = nullable)]
pub initial_turns_page: Option<ThreadResumeInitialTurnsPageParams>,
/// Deprecated and ignored by app-server. Kept only so older clients can
/// continue sending the field while rollout persistence always uses the
/// limited history policy.
@@ -435,6 +440,44 @@ pub struct ThreadResumeResponse {
#[serde(default)]
pub active_permission_profile: Option<ActivePermissionProfile>,
pub reasoning_effort: Option<ReasoningEffort>,
/// `thread/turns/list` page returned when requested by `initialTurnsPage`.
#[experimental("thread/resume.initialTurnsPage")]
#[serde(default)]
pub initial_turns_page: Option<TurnsPage>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadResumeInitialTurnsPageParams {
/// Optional turn page size.
#[ts(optional = nullable)]
pub limit: Option<u32>,
/// Optional turn pagination direction; defaults to descending.
#[ts(optional = nullable)]
pub sort_direction: Option<SortDirection>,
/// How much item detail to include for each returned turn; defaults to summary.
#[ts(optional = nullable)]
pub items_view: Option<TurnItemsView>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct TurnsPage {
pub data: Vec<Turn>,
pub next_cursor: Option<String>,
pub backwards_cursor: Option<String>,
}
impl From<ThreadTurnsListResponse> for TurnsPage {
fn from(response: ThreadTurnsListResponse) -> Self {
Self {
data: response.data,
next_cursor: response.next_cursor,
backwards_cursor: response.backwards_cursor,
}
}
}
#[derive(
+20
View File
@@ -281,6 +281,8 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev
By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. In that mode the server also skips replaying restored `thread/tokenUsage/updated`, which avoids rebuilding turns just to attribute historical usage.
Experimental clients that want the live resume subscription plus a turns page in one round trip can pass `initialTurnsPage`. It accepts the same `limit`, `sortDirection`, and `itemsView` controls as `thread/turns/list`; omitted controls use its defaults. The response includes `initialTurnsPage` with `nextCursor` and `backwardsCursor` for follow-up pagination.
By default, resume uses the latest persisted `model` and `reasoningEffort` values associated with the thread. Supplying any of `model`, `modelProvider`, `config.model`, or `config.model_reasoning_effort` disables that persisted fallback and uses the explicit overrides plus normal config resolution instead.
Example:
@@ -297,6 +299,24 @@ Example:
"excludeTurns": true
} }
{ "id": 12, "result": { "thread": { "id": "thr_123", "turns": [], … } } }
{ "method": "thread/resume", "id": 13, "params": {
"threadId": "thr_123",
"excludeTurns": true,
"initialTurnsPage": {
"limit": 20,
"sortDirection": "desc",
"itemsView": "summary"
}
} }
{ "id": 13, "result": {
"thread": { "id": "thr_123", "turns": [], … },
"initialTurnsPage": {
"data": [ ... ],
"nextCursor": "older-turns-cursor-or-null",
"backwardsCursor": "newer-turns-cursor-or-null"
}
} }
```
To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it. The returned `thread.sessionId` identifies the current live session tree root. Root threads use their own `thread.id` as `thread.sessionId`; stored threads that are not loaded also report their own `thread.id`, because resuming one makes it the root of a new live session tree. When the source history includes persisted token usage, the server also emits `thread/tokenUsage/updated` for the new thread immediately after the response. If the source thread is actively running, the fork snapshots it as if the current turn had been interrupted first. Pass `ephemeral: true` when the fork should stay in-memory only:
@@ -213,6 +213,7 @@ use codex_app_server_protocol::ThreadRealtimeStartResponse;
use codex_app_server_protocol::ThreadRealtimeStartTransport;
use codex_app_server_protocol::ThreadRealtimeStopParams;
use codex_app_server_protocol::ThreadRealtimeStopResponse;
use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRollbackParams;
@@ -565,8 +565,28 @@ pub(super) async fn handle_pending_thread_resume_request(
has_live_in_progress_turn,
);
let token_usage_thread = pending.include_turns.then(|| thread.clone());
let mut initial_turns_page = if let Some(params) = pending.initial_turns_page.as_ref() {
match super::thread_processor::build_thread_resume_initial_turns_page(
&pending.history_items,
thread.status.clone(),
has_live_in_progress_turn,
active_turn,
params,
) {
Ok(page) => Some(page),
Err(error) => {
outgoing.send_error(request_id, error).await;
return;
}
}
} else {
None
};
if pending.redact_resume_payloads {
redact_thread_resume_payloads(&mut thread);
redact_thread_resume_payloads(&mut thread.turns);
if let Some(initial_turns_page) = initial_turns_page.as_mut() {
redact_thread_resume_payloads(&mut initial_turns_page.data);
}
}
{
@@ -635,6 +655,7 @@ pub(super) async fn handle_pending_thread_resume_request(
sandbox,
active_permission_profile,
reasoning_effort,
initial_turns_page,
};
outgoing.send_response(request_id, response).await;
// Match cold resume: metadata-only resume should attach the listener without
@@ -2241,8 +2241,6 @@ impl ThreadRequestProcessor {
sort_direction,
items_view,
} = params;
let items_view = items_view.unwrap_or(TurnItemsView::Summary);
let thread_uuid = ThreadId::from_string(&thread_id)
.map_err(|err| invalid_request(format!("invalid thread id: {err}")))?;
@@ -2270,60 +2268,20 @@ impl ThreadRequestProcessor {
} else {
None
};
let mut turns = reconstruct_thread_turns_for_turns_list(
build_thread_turns_page_response(
&items,
self.thread_watch_manager
.loaded_status_for_thread(&thread_uuid.to_string())
.await,
has_live_running_thread,
active_turn,
);
for turn in &mut turns {
match items_view {
TurnItemsView::NotLoaded => {
turn.items.clear();
turn.items_view = TurnItemsView::NotLoaded;
}
TurnItemsView::Summary => {
let first_user_message = turn
.items
.iter()
.find(|item| matches!(item, ThreadItem::UserMessage { .. }))
.cloned();
let final_agent_message = turn
.items
.iter()
.rev()
.find(|item| matches!(item, ThreadItem::AgentMessage { .. }))
.cloned();
turn.items = match (first_user_message, final_agent_message) {
(Some(user_message), Some(agent_message))
if user_message.id() != agent_message.id() =>
{
vec![user_message, agent_message]
}
(Some(user_message), _) => vec![user_message],
(None, Some(agent_message)) => vec![agent_message],
(None, None) => Vec::new(),
};
turn.items_view = TurnItemsView::Summary;
}
TurnItemsView::Full => {
turn.items_view = TurnItemsView::Full;
}
}
}
let page = paginate_thread_turns(
turns,
cursor.as_deref(),
limit,
sort_direction.unwrap_or(SortDirection::Desc),
)?;
Ok(ThreadTurnsListResponse {
data: page.turns,
next_cursor: page.next_cursor,
backwards_cursor: page.backwards_cursor,
})
ThreadTurnsPageOptions {
cursor: cursor.as_deref(),
limit,
sort_direction: sort_direction.unwrap_or(SortDirection::Desc),
items_view: items_view.unwrap_or(TurnItemsView::Summary),
},
)
}
async fn load_thread_turns_list_history(
@@ -2531,6 +2489,7 @@ impl ThreadRequestProcessor {
developer_instructions,
personality,
exclude_turns,
initial_turns_page,
persist_extended_history: _persist_extended_history,
} = params;
let include_turns = !exclude_turns;
@@ -2685,8 +2644,28 @@ impl ThreadRequestProcessor {
config_snapshot.active_permission_profile,
);
let token_usage_thread = include_turns.then(|| thread.clone());
let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() {
match build_thread_resume_initial_turns_page(
&response_history.get_rollout_items(),
thread.status.clone(),
/*has_live_running_thread*/ false,
/*active_turn*/ None,
params,
) {
Ok(page) => Some(page),
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
}
}
} else {
None
};
if redact_resume_payloads {
redact_thread_resume_payloads(&mut thread);
redact_thread_resume_payloads(&mut thread.turns);
if let Some(initial_turns_page) = initial_turns_page.as_mut() {
redact_thread_resume_payloads(&mut initial_turns_page.data);
}
}
let response = ThreadResumeResponse {
@@ -2702,6 +2681,7 @@ impl ThreadRequestProcessor {
sandbox,
active_permission_profile,
reasoning_effort: session_configured.reasoning_effort,
initial_turns_page,
};
let connection_id = request_id.connection_id;
@@ -2926,6 +2906,7 @@ impl ThreadRequestProcessor {
emit_thread_goal_update,
thread_goal_state_db,
include_turns: !params.exclude_turns,
initial_turns_page: params.initial_turns_page.clone(),
redact_resume_payloads,
}),
);
@@ -3706,6 +3687,95 @@ fn parse_thread_turns_cursor(cursor: &str) -> Result<ThreadTurnsCursor, JSONRPCE
serde_json::from_str(cursor).map_err(|_| invalid_request(format!("invalid cursor: {cursor}")))
}
struct ThreadTurnsPageOptions<'a> {
cursor: Option<&'a str>,
limit: Option<u32>,
sort_direction: SortDirection,
items_view: TurnItemsView,
}
fn build_thread_turns_page_response(
items: &[RolloutItem],
loaded_status: ThreadStatus,
has_live_running_thread: bool,
active_turn: Option<Turn>,
options: ThreadTurnsPageOptions<'_>,
) -> Result<ThreadTurnsListResponse, JSONRPCErrorError> {
let mut turns = reconstruct_thread_turns_for_turns_list(
items,
loaded_status,
has_live_running_thread,
active_turn,
);
apply_thread_turns_items_view(&mut turns, options.items_view);
let page = paginate_thread_turns(turns, options.cursor, options.limit, options.sort_direction)?;
Ok(ThreadTurnsListResponse {
data: page.turns,
next_cursor: page.next_cursor,
backwards_cursor: page.backwards_cursor,
})
}
pub(super) fn build_thread_resume_initial_turns_page(
items: &[RolloutItem],
loaded_status: ThreadStatus,
has_live_running_thread: bool,
active_turn: Option<Turn>,
params: &ThreadResumeInitialTurnsPageParams,
) -> Result<codex_app_server_protocol::TurnsPage, JSONRPCErrorError> {
build_thread_turns_page_response(
items,
loaded_status,
has_live_running_thread,
active_turn,
ThreadTurnsPageOptions {
cursor: None,
limit: params.limit,
sort_direction: params.sort_direction.unwrap_or(SortDirection::Desc),
items_view: params.items_view.unwrap_or(TurnItemsView::Summary),
},
)
.map(Into::into)
}
fn apply_thread_turns_items_view(turns: &mut [Turn], items_view: TurnItemsView) {
for turn in turns {
match items_view {
TurnItemsView::NotLoaded => {
turn.items.clear();
turn.items_view = TurnItemsView::NotLoaded;
}
TurnItemsView::Summary => {
let first_user_message = turn
.items
.iter()
.find(|item| matches!(item, ThreadItem::UserMessage { .. }))
.cloned();
let final_agent_message = turn
.items
.iter()
.rev()
.find(|item| matches!(item, ThreadItem::AgentMessage { .. }))
.cloned();
turn.items = match (first_user_message, final_agent_message) {
(Some(user_message), Some(agent_message))
if user_message.id() != agent_message.id() =>
{
vec![user_message, agent_message]
}
(Some(user_message), _) => vec![user_message],
(None, Some(agent_message)) => vec![agent_message],
(None, None) => Vec::new(),
};
turn.items_view = TurnItemsView::Summary;
}
TurnItemsView::Full => {
turn.items_view = TurnItemsView::Full;
}
}
}
}
fn reconstruct_thread_turns_for_turns_list(
items: &[RolloutItem],
loaded_status: ThreadStatus,
@@ -652,6 +652,7 @@ mod thread_processor_behavior_tests {
developer_instructions: None,
personality: None,
exclude_turns: false,
initial_turns_page: None,
persist_extended_history: false,
};
let config_snapshot = ThreadConfigSnapshot {
@@ -1,6 +1,6 @@
use codex_app_server_protocol::McpToolCallResult;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::Turn;
use serde_json::Value as JsonValue;
// Temporary bandaid for remote clients: thread/resume can include large MCP and
@@ -14,8 +14,8 @@ pub(super) fn should_redact_thread_resume_payloads(client_name: Option<&str>) ->
client_name.is_some_and(|client_name| CHATGPT_REMOTE_CLIENT_NAMES.contains(&client_name))
}
pub(super) fn redact_thread_resume_payloads(thread: &mut Thread) {
for turn in &mut thread.turns {
pub(super) fn redact_thread_resume_payloads(turns: &mut [Turn]) {
for turn in turns {
turn.items.retain_mut(|item| match item {
ThreadItem::McpToolCall {
arguments,
@@ -55,8 +55,8 @@ mod tests {
use codex_app_server_protocol::McpToolCallError;
use codex_app_server_protocol::McpToolCallStatus;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStatus;
use codex_utils_absolute_path::test_support::PathBufExt;
@@ -100,7 +100,7 @@ mod tests {
},
]);
redact_thread_resume_payloads(&mut thread);
redact_thread_resume_payloads(&mut thread.turns);
assert_eq!(thread.turns[0].items.len(), 2);
assert_eq!(
@@ -146,7 +146,7 @@ mod tests {
duration_ms: Some(8),
}]);
redact_thread_resume_payloads(&mut thread);
redact_thread_resume_payloads(&mut thread.turns);
assert_eq!(
thread.turns[0].items[0],
+2
View File
@@ -35,6 +35,8 @@ pub(crate) struct PendingThreadResumeRequest {
pub(crate) emit_thread_goal_update: bool,
pub(crate) thread_goal_state_db: Option<StateDbHandle>,
pub(crate) include_turns: bool,
pub(crate) initial_turns_page:
Option<codex_app_server_protocol::ThreadResumeInitialTurnsPageParams>,
pub(crate) redact_resume_payloads: bool,
}
@@ -24,6 +24,7 @@ use codex_app_server_protocol::ThreadListResponse;
use codex_app_server_protocol::ThreadNameUpdatedNotification;
use codex_app_server_protocol::ThreadReadParams;
use codex_app_server_protocol::ThreadReadResponse;
use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadSetNameParams;
@@ -631,6 +632,77 @@ async fn thread_read_can_return_archived_threads_by_id() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_resume_initial_turns_page_matches_requested_turns_list_page() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let filename_ts = "2025-01-05T12-00-00";
let conversation_id = create_fake_rollout_with_text_elements(
codex_home.path(),
filename_ts,
"2025-01-05T12:00:00Z",
"first",
vec![],
Some("mock_provider"),
/*git_info*/ None,
)?;
let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
append_user_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "second")?;
append_user_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "third")?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let turns_list_id = mcp
.send_thread_turns_list_request(ThreadTurnsListParams {
thread_id: conversation_id.clone(),
cursor: None,
limit: Some(2),
sort_direction: Some(SortDirection::Asc),
items_view: Some(TurnItemsView::NotLoaded),
})
.await?;
let turns_list_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turns_list_id)),
)
.await??;
let expected_page = to_response::<ThreadTurnsListResponse>(turns_list_resp)?;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
exclude_turns: true,
initial_turns_page: Some(ThreadResumeInitialTurnsPageParams {
limit: Some(2),
sort_direction: Some(SortDirection::Asc),
items_view: Some(TurnItemsView::NotLoaded),
}),
..Default::default()
})
.await?;
let resume_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse {
thread,
initial_turns_page,
..
} = to_response::<ThreadResumeResponse>(resume_resp)?;
assert!(thread.turns.is_empty());
assert_eq!(
initial_turns_page,
Some(codex_app_server_protocol::TurnsPage::from(expected_page))
);
Ok(())
}
#[tokio::test]
async fn thread_turns_list_rejects_cursor_when_anchor_turn_is_rolled_back() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -37,6 +37,7 @@ use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams;
use codex_app_server_protocol::ThreadMetadataUpdateParams;
use codex_app_server_protocol::ThreadReadParams;
use codex_app_server_protocol::ThreadReadResponse;
use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadSource;
@@ -530,48 +531,59 @@ async fn thread_resume_returns_rollout_history() -> Result<()> {
#[tokio::test]
async fn thread_resume_redacts_payloads_for_chatgpt_remote_clients() -> Result<()> {
for client_name in ["codex_chatgpt_android_remote", "codex_chatgpt_ios_remote"] {
let remote_thread = resume_redaction_fixture(Some(client_name)).await?;
let remote_turn = remote_thread
let remote_resume = resume_redaction_fixture(Some(client_name)).await?;
let remote_turn = remote_resume
.thread
.turns
.first()
.expect("remote resume should include a turn");
let remote_mcp_item = remote_turn
.items
.iter()
.find(|item| matches!(item, ThreadItem::McpToolCall { .. }))
.expect("remote resume should include redacted MCP item");
let ThreadItem::McpToolCall {
arguments,
result,
error,
..
} = remote_mcp_item
else {
unreachable!("matched MCP item");
};
assert_eq!(arguments, &json!("[redacted]"));
let result = result.as_ref().expect("redacted MCP result");
assert_eq!(
result.content,
vec![json!({
"type": "text",
"text": "[redacted]",
})]
);
assert_eq!(result.structured_content, None);
assert_eq!(result.meta, None);
assert_eq!(error, &None);
assert!(
!remote_turn
let remote_page_turn = remote_resume
.initial_turns_page
.as_ref()
.expect("remote resume should include the requested initial turns page")
.data
.first()
.expect("remote initial turns page should include a turn");
for remote_turn in [remote_turn, remote_page_turn] {
let remote_mcp_item = remote_turn
.items
.iter()
.any(|item| matches!(item, ThreadItem::ImageGeneration { .. })),
"remote resume should drop image generation items for {client_name}"
);
.find(|item| matches!(item, ThreadItem::McpToolCall { .. }))
.expect("remote resume should include redacted MCP item");
let ThreadItem::McpToolCall {
arguments,
result,
error,
..
} = remote_mcp_item
else {
unreachable!("matched MCP item");
};
assert_eq!(arguments, &json!("[redacted]"));
let result = result.as_ref().expect("redacted MCP result");
assert_eq!(
result.content,
vec![json!({
"type": "text",
"text": "[redacted]",
})]
);
assert_eq!(result.structured_content, None);
assert_eq!(result.meta, None);
assert_eq!(error, &None);
assert!(
!remote_turn
.items
.iter()
.any(|item| matches!(item, ThreadItem::ImageGeneration { .. })),
"remote resume should drop image generation items for {client_name}"
);
}
}
let normal_thread = resume_redaction_fixture(Some("some_other_client")).await?;
let normal_turn = normal_thread
let normal_resume = resume_redaction_fixture(Some("some_other_client")).await?;
let normal_turn = normal_resume
.thread
.turns
.first()
.expect("normal resume should include a turn");
@@ -616,9 +628,7 @@ async fn thread_resume_redacts_payloads_for_chatgpt_remote_clients() -> Result<(
Ok(())
}
async fn resume_redaction_fixture(
client_name: Option<&str>,
) -> Result<codex_app_server_protocol::Thread> {
async fn resume_redaction_fixture(client_name: Option<&str>) -> Result<ThreadResumeResponse> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
@@ -658,6 +668,11 @@ async fn resume_redaction_fixture(
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
initial_turns_page: Some(ThreadResumeInitialTurnsPageParams {
limit: None,
sort_direction: None,
items_view: Some(TurnItemsView::Full),
}),
..Default::default()
})
.await?;
@@ -666,8 +681,7 @@ async fn resume_redaction_fixture(
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
Ok(thread)
to_response::<ThreadResumeResponse>(resume_resp)
}
fn append_resume_redaction_history(
@@ -2448,11 +2462,13 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R
..Default::default()
})
.await?;
timeout(
let running_turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_response_message(RequestId::Integer(running_turn_id)),
)
.await??;
let TurnStartResponse { turn: running_turn } =
to_response::<TurnStartResponse>(running_turn_resp)?;
timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_notification_message("turn/started"),
@@ -2464,6 +2480,11 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R
thread_id: thread.id.clone(),
model: Some("not-the-running-model".to_string()),
cwd: Some("/tmp".to_string()),
initial_turns_page: Some(ThreadResumeInitialTurnsPageParams {
limit: None,
sort_direction: None,
items_view: None,
}),
..Default::default()
})
.await?;
@@ -2472,9 +2493,23 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R
primary.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse { thread, model, .. } =
to_response::<ThreadResumeResponse>(resume_resp)?;
let ThreadResumeResponse {
thread,
model,
initial_turns_page,
..
} = to_response::<ThreadResumeResponse>(resume_resp)?;
assert_eq!(model, "gpt-5.4");
let initial_turns_page = initial_turns_page.expect("resume should include initial turns page");
let resumed_running_turn = initial_turns_page
.data
.first()
.expect("resume page should include the running turn");
assert_eq!(resumed_running_turn.id, running_turn.id);
assert_eq!(resumed_running_turn.items_view, TurnItemsView::Summary);
assert_eq!(resumed_running_turn.status, TurnStatus::InProgress);
assert!(initial_turns_page.backwards_cursor.is_some());
assert_eq!(initial_turns_page.next_cursor, None);
// The running-thread resume response is queued onto the thread listener task.
// If the in-flight turn completes before that queued command runs, the response
// can legitimately observe the thread as idle.
@@ -0,0 +1,20 @@
load("//:defs.bzl", "codex_rust_crate", "workspace_root_test")
codex_rust_crate(
name = "app-server-start-bench",
crate_name = "codex_app_server_start_bench",
)
workspace_root_test(
name = "startup-bench",
# Include the Bazel-built CLI in this target's runfiles and expose its
# resolved executable path through the environment consumed by the runner.
runfile_env = {
"//codex-rs/cli:codex": "CODEX_BIN",
},
size = "large",
tags = ["manual"],
test_bin = ":codex-app-server-start-bench",
timeout = "long",
workspace_root_marker = "//codex-rs/utils/cargo-bin:repo_root.marker",
)
@@ -0,0 +1,16 @@
[package]
name = "codex-app-server-start-bench"
version.workspace = true
edition.workspace = true
license.workspace = true
publish = false
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
codex-app-server-protocol = { workspace = true }
serde_json = { workspace = true }
divan = { workspace = true }
tempfile = { workspace = true }
@@ -0,0 +1,174 @@
use std::io::BufRead;
use std::io::BufReader;
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use std::process::Child;
use std::process::ChildStdin;
use std::process::ChildStdout;
use std::process::Command;
use std::process::Stdio;
use anyhow::Context;
use anyhow::Result;
use anyhow::bail;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::InitializeResponse;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use divan::Bencher;
fn main() {
divan::main();
}
// Process startup is slow enough that 30 samples keep runs practical, and
// sample size 1 lets each measured server be reaped before the next starts.
// The benchmark warms one CODEX_HOME before timing to measure normal restarts.
// This e2e runner receives its separately built Codex binary from just or Bazel.
#[divan::bench(sample_count = 30, sample_size = 1, skip_ext_time)]
#[allow(clippy::expect_used)]
fn initialize_response(bencher: Bencher) {
let codex_bin = std::env::var_os("CODEX_BIN")
.map(PathBuf::from)
.expect("CODEX_BIN must point to the codex binary; run via `just bench-e2e` or Bazel");
let codex_home = tempfile::tempdir().expect("benchmark CODEX_HOME should be created");
drop(
start_until_initialize_response(&codex_bin, codex_home.path())
.expect("benchmark CODEX_HOME should be initialized"),
);
bencher.bench_local(|| {
start_until_initialize_response(&codex_bin, codex_home.path())
.expect("codex app-server should return an initialize response")
});
}
/// A running app-server that has returned a valid `initialize` response.
///
/// Divan drops benchmark outputs after it stops each timer interval. Returning
/// this value from the measured closure keeps the `initialized` notification
/// and forced process reaping outside startup latency.
struct InitializedAppServer {
child: Child,
stdin: Option<ChildStdin>,
stdout: Option<BufReader<ChildStdout>>,
acknowledge_on_drop: bool,
}
/// Spawn a stdio app-server and return once it responds successfully to `initialize`.
fn start_until_initialize_response(
codex_bin: &Path,
codex_home: &Path,
) -> Result<InitializedAppServer> {
let request_id = RequestId::Integer(0);
let child = Command::new(codex_bin)
.arg("app-server")
.arg("--listen")
.arg("stdio://")
.env("CODEX_HOME", codex_home)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.with_context(|| format!("failed to spawn `{}` app-server", codex_bin.display()))?;
let mut server = InitializedAppServer {
child,
stdin: None,
stdout: None,
acknowledge_on_drop: false,
};
server.stdin = Some(
server
.child
.stdin
.take()
.context("app-server stdin unavailable")?,
);
server.stdout = Some(BufReader::new(
server
.child
.stdout
.take()
.context("app-server stdout unavailable")?,
));
let request = ClientRequest::Initialize {
request_id: request_id.clone(),
params: InitializeParams {
client_info: ClientInfo {
name: "codex-app-server-start-bench".to_string(),
title: Some("Codex App Server Start Benchmark".to_string()),
version: env!("CARGO_PKG_VERSION").to_string(),
},
capabilities: Some(InitializeCapabilities {
experimental_api: false,
request_attestation: false,
opt_out_notification_methods: None,
}),
},
};
let stdin = server
.stdin
.as_mut()
.context("app-server stdin unavailable")?;
writeln!(stdin, "{}", serde_json::to_string(&request)?)?;
stdin
.flush()
.context("failed to flush initialize request")?;
let mut line = String::new();
loop {
line.clear();
if server
.stdout
.as_mut()
.context("app-server stdout unavailable")?
.read_line(&mut line)?
== 0
{
bail!("app-server closed stdout before returning initialize response");
}
match serde_json::from_str::<JSONRPCMessage>(line.trim())? {
JSONRPCMessage::Response(JSONRPCResponse { id, result }) if id == request_id => {
let _: InitializeResponse = serde_json::from_value(result)
.context("initialize response missing expected payload")?;
server.acknowledge_on_drop = true;
return Ok(server);
}
JSONRPCMessage::Error(error) if error.id == request_id => {
bail!("initialize failed: {error:?}");
}
JSONRPCMessage::Request(_)
| JSONRPCMessage::Response(_)
| JSONRPCMessage::Notification(_)
| JSONRPCMessage::Error(_) => {}
}
}
}
impl Drop for InitializedAppServer {
fn drop(&mut self) {
if self.acknowledge_on_drop
&& let Some(stdin) = self.stdin.as_mut()
{
let initialized = JSONRPCMessage::Notification(JSONRPCNotification {
method: "initialized".to_string(),
params: None,
});
if let Ok(payload) = serde_json::to_string(&initialized) {
let _ = writeln!(stdin, "{payload}");
let _ = stdin.flush();
}
}
let _ = self.stdin.take();
let _ = self.child.kill();
let _ = self.child.wait();
}
}
@@ -1,9 +1,11 @@
use crate::session::turn_context::TurnContext;
use crate::session::turn_context::TurnEnvironment;
use crate::shell::Shell;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::TurnContextNetworkItem;
use codex_utils_absolute_path::AbsolutePathBuf;
use std::path::Path;
use super::ContextualUserFragment;
@@ -13,6 +15,7 @@ pub(crate) struct EnvironmentContext {
pub(crate) current_date: Option<String>,
pub(crate) timezone: Option<String>,
pub(crate) network: Option<NetworkContext>,
pub(crate) filesystem: Option<FileSystemContext>,
pub(crate) subagents: Option<String>,
}
@@ -83,6 +86,55 @@ impl EnvironmentContextEnvironments {
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub(crate) struct FileSystemContext {
unreadable_roots: Vec<String>,
unreadable_globs: Vec<String>,
}
impl FileSystemContext {
fn new(unreadable_roots: Vec<String>, unreadable_globs: Vec<String>) -> Option<Self> {
if unreadable_roots.is_empty() && unreadable_globs.is_empty() {
None
} else {
Some(Self {
unreadable_roots,
unreadable_globs,
})
}
}
fn from_permission_profile(permission_profile: &PermissionProfile, cwd: &Path) -> Option<Self> {
let file_system_sandbox_policy = permission_profile.file_system_sandbox_policy();
Self::new(
file_system_sandbox_policy
.get_unreadable_roots_with_cwd(cwd)
.into_iter()
.map(|root| root.to_string_lossy().into_owned())
.collect(),
file_system_sandbox_policy.get_unreadable_globs_with_cwd(cwd),
)
}
fn render(&self) -> String {
let mut rendered = "<filesystem><deny_read escalatable=\"false\">".to_string();
Self::push_rendered_element(&mut rendered, "paths", &self.unreadable_roots);
Self::push_rendered_element(&mut rendered, "globs", &self.unreadable_globs);
rendered.push_str("</deny_read></filesystem>");
rendered
}
fn push_rendered_element(rendered_filesystem: &mut String, name: &str, values: &[String]) {
if values.is_empty() {
return;
}
rendered_filesystem.push_str(&format!("<{name}>"));
rendered_filesystem.push_str(&values.join(","));
rendered_filesystem.push_str(&format!("</{name}>"));
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub(crate) struct NetworkContext {
allowed_domains: Vec<String>,
@@ -129,6 +181,7 @@ impl EnvironmentContext {
current_date,
timezone,
network,
filesystem: None,
subagents,
}
}
@@ -138,6 +191,7 @@ impl EnvironmentContext {
current_date: Option<String>,
timezone: Option<String>,
network: Option<NetworkContext>,
filesystem: Option<FileSystemContext>,
subagents: Option<String>,
) -> Self {
Self {
@@ -145,6 +199,7 @@ impl EnvironmentContext {
current_date,
timezone,
network,
filesystem,
subagents,
}
}
@@ -157,6 +212,7 @@ impl EnvironmentContext {
&& self.current_date == other.current_date
&& self.timezone == other.timezone
&& self.network == other.network
&& self.filesystem == other.filesystem
&& self.subagents == other.subagents
}
@@ -165,6 +221,7 @@ impl EnvironmentContext {
after: &EnvironmentContext,
) -> Self {
let before_network = Self::network_from_turn_context_item(before);
let before_filesystem = Self::filesystem_from_turn_context_item(before);
let environments = match &after.environments {
EnvironmentContextEnvironments::Single(environment) => {
if before.cwd.as_path() != environment.cwd.as_path() {
@@ -186,17 +243,25 @@ impl EnvironmentContext {
} else {
before_network
};
let filesystem = if before_filesystem != after.filesystem {
after.filesystem.clone()
} else {
before_filesystem
};
EnvironmentContext::new_with_environments(
environments,
after.current_date.clone(),
after.timezone.clone(),
network,
filesystem,
/*subagents*/ None,
)
}
pub(crate) fn from_turn_context(turn_context: &TurnContext, shell: &Shell) -> Self {
Self::new(
#[allow(deprecated)]
let cwd = &turn_context.cwd;
let mut context = Self::new(
EnvironmentContextEnvironment::from_turn_environments(
&turn_context.environments.turn_environments,
shell,
@@ -205,7 +270,10 @@ impl EnvironmentContext {
turn_context.timezone.clone(),
Self::network_from_turn_context(turn_context),
/*subagents*/ None,
)
);
context.filesystem =
FileSystemContext::from_permission_profile(&turn_context.permission_profile, cwd);
context
}
pub(crate) fn from_turn_context_item(
@@ -216,11 +284,14 @@ impl EnvironmentContext {
Ok(cwd) => cwd,
Err(_) => AbsolutePathBuf::resolve_path_against_base(&turn_context_item.cwd, "/"),
};
Self::new(
vec![EnvironmentContextEnvironment::legacy(cwd, shell)],
Self::new_with_environments(
EnvironmentContextEnvironments::from_vec(vec![EnvironmentContextEnvironment::legacy(
cwd, shell,
)]),
turn_context_item.current_date.clone(),
turn_context_item.timezone.clone(),
Self::network_from_turn_context_item(turn_context_item),
Self::filesystem_from_turn_context_item(turn_context_item),
/*subagents*/ None,
)
}
@@ -266,6 +337,15 @@ impl EnvironmentContext {
denied_domains.clone(),
))
}
fn filesystem_from_turn_context_item(
turn_context_item: &TurnContextItem,
) -> Option<FileSystemContext> {
FileSystemContext::from_permission_profile(
&turn_context_item.permission_profile(),
&turn_context_item.cwd,
)
}
}
impl ContextualUserFragment for EnvironmentContext {
@@ -324,6 +404,9 @@ impl ContextualUserFragment for EnvironmentContext {
// lines.push(" <network enabled=\"false\" />".to_string());
}
}
if let Some(filesystem) = &self.filesystem {
lines.push(format!(" {}", filesystem.render()));
}
if let Some(subagents) = &self.subagents {
lines.push(" <subagents>".to_string());
lines.extend(subagents.lines().map(|line| format!(" {line}")));
@@ -79,6 +79,36 @@ fn serialize_environment_context_with_network() {
assert_eq!(context.render(), expected);
}
#[test]
fn serialize_environment_context_with_filesystem_denied_reads() {
let mut context = EnvironmentContext::new(
vec![EnvironmentContextEnvironment {
id: "local".to_string(),
cwd: test_path_buf("/repo").abs(),
shell: fake_shell_name(),
}],
/*current_date*/ None,
/*timezone*/ None,
/*network*/ None,
/*subagents*/ None,
);
context.filesystem = FileSystemContext::new(
vec!["/repo/private".to_string()],
vec!["/repo/private/**".to_string()],
);
let expected = format!(
r#"<environment_context>
<cwd>{}</cwd>
<shell>bash</shell>
<filesystem><deny_read escalatable="false"><paths>/repo/private</paths><globs>/repo/private/**</globs></deny_read></filesystem>
</environment_context>"#,
test_path_buf("/repo").display()
);
assert_eq!(context.render(), expected);
}
#[test]
fn serialize_read_only_environment_context() {
let context = EnvironmentContext::new(
@@ -4,6 +4,7 @@ use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::SandboxMode;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::format_allow_prefixes;
use codex_protocol::permissions::FileSystemSandboxPolicy;
use codex_protocol::permissions::NetworkSandboxPolicy;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::GranularApprovalConfig;
@@ -68,9 +69,11 @@ impl PermissionsInstructions {
exec_permission_approvals_enabled: bool,
request_permissions_tool_enabled: bool,
) -> Self {
let (sandbox_mode, writable_roots) = sandbox_prompt_from_profile(permission_profile, cwd);
let file_system_sandbox_policy = permission_profile.file_system_sandbox_policy();
let (sandbox_mode, writable_roots) =
sandbox_prompt_from_policy(&file_system_sandbox_policy, cwd);
Self::from_permissions_with_network(
Self::from_permissions_with_network_and_denied_reads(
sandbox_mode,
network_access_from_policy(permission_profile.network_sandbox_policy()),
PermissionsPromptConfig {
@@ -81,14 +84,32 @@ impl PermissionsInstructions {
request_permissions_tool_enabled,
},
writable_roots,
denied_reads_text(&file_system_sandbox_policy, cwd),
)
}
#[cfg(test)]
fn from_permissions_with_network(
sandbox_mode: SandboxMode,
network_access: NetworkAccess,
config: PermissionsPromptConfig<'_>,
writable_roots: Option<Vec<WritableRoot>>,
) -> Self {
Self::from_permissions_with_network_and_denied_reads(
sandbox_mode,
network_access,
config,
writable_roots,
/*denied_reads*/ None,
)
}
fn from_permissions_with_network_and_denied_reads(
sandbox_mode: SandboxMode,
network_access: NetworkAccess,
config: PermissionsPromptConfig<'_>,
writable_roots: Option<Vec<WritableRoot>>,
denied_reads: Option<String>,
) -> Self {
let mut text = String::new();
append_section(&mut text, &sandbox_text(sandbox_mode, network_access));
@@ -105,6 +126,9 @@ impl PermissionsInstructions {
if let Some(writable_roots) = writable_roots_text(writable_roots) {
append_section(&mut text, &writable_roots);
}
if let Some(denied_reads) = denied_reads {
append_section(&mut text, &denied_reads);
}
if !text.ends_with('\n') {
text.push('\n');
}
@@ -112,27 +136,19 @@ impl PermissionsInstructions {
}
}
fn sandbox_prompt_from_profile(
permission_profile: &PermissionProfile,
fn sandbox_prompt_from_policy(
file_system_policy: &FileSystemSandboxPolicy,
cwd: &Path,
) -> (SandboxMode, Option<Vec<WritableRoot>>) {
match permission_profile {
PermissionProfile::Disabled | PermissionProfile::External { .. } => {
(SandboxMode::DangerFullAccess, None)
}
PermissionProfile::Managed { .. } => {
let file_system_policy = permission_profile.file_system_sandbox_policy();
if file_system_policy.has_full_disk_write_access() {
return (SandboxMode::DangerFullAccess, None);
}
if file_system_policy.has_full_disk_write_access() {
return (SandboxMode::DangerFullAccess, None);
}
let writable_roots = file_system_policy.get_writable_roots_with_cwd(cwd);
if writable_roots.is_empty() {
(SandboxMode::ReadOnly, None)
} else {
(SandboxMode::WorkspaceWrite, Some(writable_roots))
}
}
let writable_roots = file_system_policy.get_writable_roots_with_cwd(cwd);
if writable_roots.is_empty() {
(SandboxMode::ReadOnly, None)
} else {
(SandboxMode::WorkspaceWrite, Some(writable_roots))
}
}
@@ -254,6 +270,28 @@ fn writable_roots_text(writable_roots: Option<Vec<WritableRoot>>) -> Option<Stri
})
}
fn denied_reads_text(file_system_policy: &FileSystemSandboxPolicy, cwd: &Path) -> Option<String> {
let mut entries = file_system_policy
.get_unreadable_roots_with_cwd(cwd)
.into_iter()
.map(|root| format!("- path `{}`", root.to_string_lossy()))
.collect::<Vec<_>>();
entries.extend(
file_system_policy
.get_unreadable_globs_with_cwd(cwd)
.into_iter()
.map(|glob| format!("- glob `{glob}`")),
);
if entries.is_empty() {
return None;
}
Some(format!(
"## Denied filesystem reads\nThe active permission profile denies reading these paths/globs. Do not request escalation or additional permissions to read them; these denials are policy restrictions.\n{}",
entries.join("\n")
))
}
fn approved_command_prefixes_text(exec_policy: &Policy) -> Option<String> {
format_allow_prefixes(exec_policy.get_allowed_prefixes())
.filter(|prefixes| !prefixes.is_empty())
@@ -83,6 +83,51 @@ fn builds_permissions_from_profile() {
assert!(text.contains(writable_root.to_string_lossy().as_ref()));
}
#[test]
fn builds_permissions_from_profile_with_denied_reads() {
let cwd = PathBuf::from("/tmp");
let denied_root =
AbsolutePathBuf::from_absolute_path(cwd.join("blocked")).expect("absolute path");
let permission_profile = PermissionProfile::from_runtime_permissions(
&FileSystemSandboxPolicy::restricted(vec![
FileSystemSandboxEntry {
path: FileSystemPath::Special {
value: codex_protocol::permissions::FileSystemSpecialPath::Root,
},
access: FileSystemAccessMode::Read,
},
FileSystemSandboxEntry {
path: FileSystemPath::Path {
path: denied_root.clone(),
},
access: FileSystemAccessMode::Deny,
},
FileSystemSandboxEntry {
path: FileSystemPath::GlobPattern {
pattern: "/tmp/blocked/**".to_string(),
},
access: FileSystemAccessMode::Deny,
},
]),
NetworkSandboxPolicy::Restricted,
);
let instructions = PermissionsInstructions::from_permission_profile(
&permission_profile,
AskForApproval::OnRequest,
ApprovalsReviewer::AutoReview,
&Policy::empty(),
&cwd,
/*exec_permission_approvals_enabled*/ false,
/*request_permissions_tool_enabled*/ false,
);
let text = instructions.body();
assert!(text.contains("## Denied filesystem reads"));
assert!(text.contains("Do not request escalation or additional permissions"));
assert!(text.contains(denied_root.to_string_lossy().as_ref()));
assert!(text.contains("glob `/tmp/blocked/**`"));
}
#[test]
fn includes_request_rule_instructions_for_on_request() {
let mut exec_policy = Policy::empty();
+2
View File
@@ -149,6 +149,8 @@ use prompt::GuardianTranscriptEntryKind;
#[cfg(test)]
use prompt::build_guardian_prompt_items;
#[cfg(test)]
use prompt::build_guardian_prompt_items_with_parent_turn;
#[cfg(test)]
use prompt::collect_guardian_transcript_entries;
#[cfg(test)]
use prompt::guardian_output_schema;
+49
View File
@@ -10,6 +10,7 @@ use serde_json::Value;
use crate::compact::content_items_to_text;
use crate::event_mapping::is_contextual_user_message_content;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use codex_utils_output_truncation::approx_bytes_for_tokens;
use codex_utils_output_truncation::approx_token_count;
use codex_utils_output_truncation::approx_tokens_from_byte_count;
@@ -86,11 +87,29 @@ pub(crate) enum GuardianPromptMode {
/// Split the variable request into separate user content items so the
/// Responses request snapshot shows clear boundaries while preserving exact
/// prompt text through trailing newlines.
#[cfg(test)]
pub(crate) async fn build_guardian_prompt_items(
session: &Session,
retry_reason: Option<String>,
request: GuardianApprovalRequest,
mode: GuardianPromptMode,
) -> serde_json::Result<GuardianPromptItems> {
build_guardian_prompt_items_with_parent_turn(
session,
/*parent_turn*/ None,
retry_reason,
request,
mode,
)
.await
}
pub(crate) async fn build_guardian_prompt_items_with_parent_turn(
session: &Session,
parent_turn: Option<&TurnContext>,
retry_reason: Option<String>,
request: GuardianApprovalRequest,
mode: GuardianPromptMode,
) -> serde_json::Result<GuardianPromptItems> {
let history = session.clone_history().await;
let transcript_entries = collect_guardian_transcript_entries(history.raw_items());
@@ -172,6 +191,11 @@ pub(crate) async fn build_guardian_prompt_items(
if let Some(note) = omission_note {
push_text(format!("\n{note}\n"));
}
if let Some(denied_reads_context) = parent_turn.and_then(parent_turn_denied_reads_context) {
push_text("\n>>> PARENT TURN PERMISSION CONTEXT START\n".to_string());
push_text(denied_reads_context);
push_text(">>> PARENT TURN PERMISSION CONTEXT END\n".to_string());
}
match &request {
GuardianApprovalRequest::NetworkAccess { trigger, .. } => {
push_text(">>> APPROVAL REQUEST START\n".to_string());
@@ -216,6 +240,31 @@ pub(crate) async fn build_guardian_prompt_items(
})
}
fn parent_turn_denied_reads_context(turn: &TurnContext) -> Option<String> {
#[allow(deprecated)]
let cwd = &turn.cwd;
let file_system_policy = turn.permission_profile.file_system_sandbox_policy();
let mut entries = file_system_policy
.get_unreadable_roots_with_cwd(cwd)
.into_iter()
.map(|root| format!("- path `{}`", root.to_string_lossy()))
.collect::<Vec<_>>();
entries.extend(
file_system_policy
.get_unreadable_globs_with_cwd(cwd)
.into_iter()
.map(|glob| format!("- glob `{glob}`")),
);
if entries.is_empty() {
return None;
}
Some(format!(
"The parent turn's active permission profile denies reading these paths/globs. These are policy restrictions; do not approve escalation whose purpose is to read them.\n{}\n",
entries.join("\n")
))
}
enum GuardianPromptShape {
Full,
Delta { already_seen_entry_count: usize },
+3 -2
View File
@@ -51,7 +51,7 @@ use super::GUARDIAN_REVIEWER_NAME;
use super::GuardianApprovalRequest;
use super::prompt::GuardianPromptMode;
use super::prompt::GuardianTranscriptCursor;
use super::prompt::build_guardian_prompt_items;
use super::prompt::build_guardian_prompt_items_with_parent_turn;
use super::prompt::guardian_policy_prompt;
use super::prompt::guardian_policy_prompt_with_config;
@@ -686,8 +686,9 @@ async fn run_review_on_session(
)
.await;
build_guardian_prompt_items(
build_guardian_prompt_items_with_parent_turn(
params.parent_session.as_ref(),
Some(params.parent_turn.as_ref()),
params.retry_reason.clone(),
params.request.clone(),
prompt_mode,
+62
View File
@@ -33,6 +33,11 @@ use codex_protocol::models::ContentItem;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::permissions::FileSystemAccessMode;
use codex_protocol::permissions::FileSystemPath;
use codex_protocol::permissions::FileSystemSandboxEntry;
use codex_protocol::permissions::FileSystemSandboxPolicy;
use codex_protocol::permissions::NetworkSandboxPolicy;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
@@ -357,6 +362,63 @@ async fn build_guardian_prompt_full_mode_preserves_initial_review_format() -> an
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn build_guardian_prompt_includes_parent_turn_denied_reads() -> anyhow::Result<()> {
let (mut session, mut turn) = crate::session::tests::make_session_and_context().await;
session.conversation_id = fixed_guardian_parent_session_id();
let denied_root = test_path_buf("/repo/private").abs();
turn.permission_profile = PermissionProfile::from_runtime_permissions(
&FileSystemSandboxPolicy::restricted(vec![
FileSystemSandboxEntry {
path: FileSystemPath::Special {
value: codex_protocol::permissions::FileSystemSpecialPath::Root,
},
access: FileSystemAccessMode::Read,
},
FileSystemSandboxEntry {
path: FileSystemPath::Path {
path: denied_root.clone(),
},
access: FileSystemAccessMode::Deny,
},
FileSystemSandboxEntry {
path: FileSystemPath::GlobPattern {
pattern: "/repo/private/**".to_string(),
},
access: FileSystemAccessMode::Deny,
},
]),
NetworkSandboxPolicy::Restricted,
);
let session = Arc::new(session);
let turn = Arc::new(turn);
seed_guardian_parent_history(&session, &turn).await;
let prompt = build_guardian_prompt_items_with_parent_turn(
session.as_ref(),
Some(turn.as_ref()),
Some("Sandbox denied reading /repo/private/secret.txt.".to_string()),
GuardianApprovalRequest::Shell {
id: "shell-1".to_string(),
command: vec!["cat".to_string(), "/repo/private/secret.txt".to_string()],
cwd: test_path_buf("/repo").abs(),
sandbox_permissions: crate::sandboxing::SandboxPermissions::RequireEscalated,
additional_permissions: None,
justification: Some("Need to inspect the secret file.".to_string()),
},
GuardianPromptMode::Full,
)
.await?;
let text = guardian_prompt_text(&prompt.items);
assert!(text.contains("PARENT TURN PERMISSION CONTEXT START"));
assert!(text.contains("do not approve escalation whose purpose is to read them"));
assert!(text.contains(denied_root.to_string_lossy().as_ref()));
assert!(text.contains("glob `/repo/private/**`"));
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn build_guardian_prompt_delta_mode_preserves_original_numbering() -> anyhow::Result<()> {
let (session, turn) = guardian_test_session_and_turn_with_base_url("http://localhost").await;
@@ -1,11 +1,18 @@
use std::sync::Arc;
use std::sync::Weak;
use codex_protocol::items::TurnItem;
use codex_tools::ConversationHistory;
use codex_tools::ExtensionTurnItem;
use codex_tools::ToolCall as ExtensionToolCall;
use codex_tools::ToolName;
use codex_tools::ToolSpec;
use codex_tools::TurnItemEmissionFuture;
use codex_tools::TurnItemEmitter;
use crate::function_tool::FunctionCallError;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolOutput;
use crate::tools::context::ToolPayload;
@@ -52,6 +59,39 @@ impl CoreToolRuntime for ExtensionToolAdapter {
}
}
struct CoreTurnItemEmitter {
session: Weak<Session>,
turn: Weak<TurnContext>,
}
fn extension_turn_item(item: ExtensionTurnItem) -> TurnItem {
match item {
ExtensionTurnItem::WebSearch(item) => TurnItem::WebSearch(item),
}
}
impl TurnItemEmitter for CoreTurnItemEmitter {
fn emit_started<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> {
Box::pin(async move {
let (Some(session), Some(turn)) = (self.session.upgrade(), self.turn.upgrade()) else {
return;
};
let item = extension_turn_item(item);
session.emit_turn_item_started(turn.as_ref(), &item).await;
})
}
fn emit_completed<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> {
Box::pin(async move {
let (Some(session), Some(turn)) = (self.session.upgrade(), self.turn.upgrade()) else {
return;
};
let item = extension_turn_item(item);
session.emit_turn_item_completed(turn.as_ref(), item).await;
})
}
}
async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall {
let conversation_history =
ConversationHistory::new(invocation.session.clone_history().await.into_raw_items());
@@ -61,6 +101,10 @@ async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall {
tool_name: invocation.tool_name.clone(),
truncation_policy: invocation.turn.truncation_policy,
conversation_history,
turn_item_emitter: Arc::new(CoreTurnItemEmitter {
session: Arc::downgrade(&invocation.session),
turn: Arc::downgrade(&invocation.turn),
}),
payload: invocation.payload.clone(),
}
}
@@ -69,8 +113,13 @@ async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall {
mod tests {
use std::sync::Arc;
use codex_protocol::items::TurnItem;
use codex_protocol::items::WebSearchItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::models::WebSearchAction;
use codex_protocol::protocol::EventMsg;
use codex_tools::ExtensionTurnItem;
use pretty_assertions::assert_eq;
use serde_json::json;
use tokio::sync::Mutex;
@@ -147,6 +196,16 @@ mod tests {
&self,
call: codex_tools::ToolCall,
) -> Result<Box<dyn codex_tools::ToolOutput>, codex_tools::FunctionCallError> {
let item = ExtensionTurnItem::WebSearch(WebSearchItem {
id: call.call_id.clone(),
query: "rust trait object".to_string(),
action: WebSearchAction::Search {
query: Some("rust trait object".to_string()),
queries: None,
},
});
call.turn_item_emitter.emit_started(item.clone()).await;
call.turn_item_emitter.emit_completed(item).await;
*self.captured_call.lock().await = Some(call);
Ok(Box::new(codex_tools::JsonToolOutput::new(
json!({ "ok": true }),
@@ -191,12 +250,14 @@ mod tests {
}
#[tokio::test]
async fn passes_turn_fields_to_extension_call() {
async fn passes_turn_fields_and_scoped_turn_item_emitter_to_extension_call() {
let captured_call = Arc::new(Mutex::new(None));
let handler = ExtensionToolAdapter::new(Arc::new(CapturingExtensionExecutor {
captured_call: Arc::clone(&captured_call),
}));
let (session, turn) = crate::session::tests::make_session_and_context().await;
let (session, turn, rx) = crate::session::tests::make_session_and_context_with_rx().await;
let weak_session = Arc::downgrade(&session);
let weak_turn = Arc::downgrade(&turn);
let turn_id = turn.sub_id.clone();
let truncation_policy = turn.truncation_policy;
let history_item = ResponseItem::Message {
@@ -211,8 +272,8 @@ mod tests {
.record_conversation_items(&turn, std::slice::from_ref(&history_item))
.await;
let invocation = ToolInvocation {
session: session.into(),
turn: turn.into(),
session,
turn,
cancellation_token: tokio_util::sync::CancellationToken::new(),
tracker: Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())),
call_id: "call-extension".to_string(),
@@ -228,6 +289,8 @@ mod tests {
.expect("extension call should succeed");
let captured_call = captured_call.lock().await.clone().expect("captured call");
assert!(weak_session.upgrade().is_none());
assert!(weak_turn.upgrade().is_none());
assert_eq!(captured_call.turn_id, turn_id);
assert_eq!(captured_call.call_id, "call-extension");
assert_eq!(
@@ -245,5 +308,43 @@ mod tests {
}
payload => panic!("expected function payload, got {payload:?}"),
}
let started = rx.recv().await.expect("item started event");
let EventMsg::ItemStarted(started) = started.msg else {
panic!("expected item started event");
};
let TurnItem::WebSearch(started_item) = started.item else {
panic!("expected web search item");
};
let begin = rx.recv().await.expect("legacy web search begin event");
let EventMsg::WebSearchBegin(begin) = begin.msg else {
panic!("expected legacy web search begin event");
};
let completed = rx.recv().await.expect("item completed event");
let EventMsg::ItemCompleted(completed) = completed.msg else {
panic!("expected item completed event");
};
let TurnItem::WebSearch(completed_item) = completed.item else {
panic!("expected web search item");
};
let end = rx.recv().await.expect("legacy web search end event");
let EventMsg::WebSearchEnd(end) = end.msg else {
panic!("expected legacy web search end event");
};
let expected = WebSearchItem {
id: "call-extension".to_string(),
query: "rust trait object".to_string(),
action: WebSearchAction::Search {
query: Some("rust trait object".to_string()),
queries: None,
},
};
assert_eq!(started_item, expected);
assert_eq!(completed_item, expected);
assert_eq!(begin.call_id, expected.id);
assert_eq!(end.call_id, expected.id);
assert_eq!(end.query, expected.query);
assert_eq!(end.action, expected.action);
}
}
+4
View File
@@ -11,8 +11,10 @@ pub use capabilities::NoopResponseItemInjector;
pub use capabilities::ResponseItemInjectionFuture;
pub use capabilities::ResponseItemInjector;
pub use codex_tools::ConversationHistory;
pub use codex_tools::ExtensionTurnItem;
pub use codex_tools::FunctionCallError;
pub use codex_tools::JsonToolOutput;
pub use codex_tools::NoopTurnItemEmitter;
pub use codex_tools::ResponsesApiTool;
pub use codex_tools::ToolCall;
pub use codex_tools::ToolExecutor;
@@ -20,6 +22,8 @@ pub use codex_tools::ToolName;
pub use codex_tools::ToolOutput;
pub use codex_tools::ToolPayload;
pub use codex_tools::ToolSpec;
pub use codex_tools::TurnItemEmissionFuture;
pub use codex_tools::TurnItemEmitter;
pub use codex_tools::parse_tool_input_schema;
pub use codex_tools::parse_tool_input_schema_without_compaction;
pub use contributors::ApprovalReviewContributor;
+13 -2
View File
@@ -22,6 +22,8 @@ use codex_extension_api::TurnStartInput;
use codex_extension_api::TurnStopInput;
use codex_otel::MetricsClient;
use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadGoalStatus;
use codex_protocol::protocol::TokenUsageInfo;
@@ -29,6 +31,7 @@ use crate::accounting::BudgetLimitedGoalDisposition;
use crate::accounting::GoalAccountingState;
use crate::events::GoalEventEmitter;
use crate::metrics::GoalMetrics;
use crate::runtime::GoalRuntimeConfig;
use crate::runtime::GoalRuntimeHandle;
use crate::spec::UPDATE_GOAL_TOOL_NAME;
use crate::steering::budget_limit_steering_item;
@@ -85,6 +88,11 @@ where
{
async fn on_thread_start(&self, input: ThreadStartInput<'_, C>) {
let enabled = (self.goals_enabled)(input.config);
let tools_available_for_thread = input.persistent_thread_state_available
&& !matches!(
input.session_source,
SessionSource::SubAgent(SubAgentSource::Review)
);
input
.thread_store
.insert(GoalExtensionConfig::from_enabled(enabled));
@@ -102,7 +110,10 @@ where
self.metrics.clone(),
self.thread_manager.clone(),
accounting_state,
enabled,
GoalRuntimeConfig {
enabled,
tools_available_for_thread,
},
)
});
runtime.set_enabled(enabled);
@@ -330,7 +341,7 @@ where
let Some(runtime) = goal_runtime_handle(thread_store) else {
return Vec::new();
};
if !runtime.is_enabled() {
if !runtime.tools_visible() {
return Vec::new();
}
+13 -2
View File
@@ -20,6 +20,11 @@ pub struct GoalRuntimeHandle {
inner: Arc<GoalRuntimeInner>,
}
pub(crate) struct GoalRuntimeConfig {
pub(crate) enabled: bool,
pub(crate) tools_available_for_thread: bool,
}
struct GoalRuntimeInner {
thread_id: ThreadId,
state_dbs: Arc<codex_state::StateRuntime>,
@@ -28,6 +33,7 @@ struct GoalRuntimeInner {
thread_manager: Weak<ThreadManager>,
accounting_state: Arc<GoalAccountingState>,
enabled: AtomicBool,
tools_available_for_thread: bool,
}
pub(crate) struct AccountedGoalProgress {
@@ -66,7 +72,7 @@ impl GoalRuntimeHandle {
metrics: GoalMetrics,
thread_manager: Weak<ThreadManager>,
accounting_state: Arc<GoalAccountingState>,
enabled: bool,
config: GoalRuntimeConfig,
) -> Self {
Self {
inner: Arc::new(GoalRuntimeInner {
@@ -76,7 +82,8 @@ impl GoalRuntimeHandle {
metrics,
thread_manager,
accounting_state,
enabled: AtomicBool::new(enabled),
enabled: AtomicBool::new(config.enabled),
tools_available_for_thread: config.tools_available_for_thread,
}),
}
}
@@ -89,6 +96,10 @@ impl GoalRuntimeHandle {
self.inner.enabled.load(Ordering::Relaxed)
}
pub(crate) fn tools_visible(&self) -> bool {
self.is_enabled() && self.inner.tools_available_for_thread
}
pub(crate) fn thread_id(&self) -> ThreadId {
self.inner.thread_id
}
@@ -8,6 +8,7 @@ use codex_extension_api::ExtensionData;
use codex_extension_api::ExtensionEventSink;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_extension_api::FunctionCallError;
use codex_extension_api::NoopTurnItemEmitter;
use codex_extension_api::ThreadResumeInput;
use codex_extension_api::ThreadStartInput;
use codex_extension_api::ToolCall;
@@ -28,6 +29,7 @@ use codex_protocol::config_types::Settings;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadGoalStatus;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
@@ -83,6 +85,38 @@ async fn installed_goal_tools_create_goal_and_fill_empty_preview() -> anyhow::Re
Ok(())
}
#[tokio::test]
async fn goal_tools_hidden_for_ephemeral_threads() -> anyhow::Result<()> {
let runtime = test_runtime().await?;
let thread_id = test_thread_id()?;
let tools = installed_tools_with_start(
runtime,
thread_id,
SessionSource::Cli,
/*persistent_thread_state_available*/ false,
)
.await;
assert_eq!(Vec::<String>::new(), tool_names(&tools));
Ok(())
}
#[tokio::test]
async fn goal_tools_hidden_for_review_subagents() -> anyhow::Result<()> {
let runtime = test_runtime().await?;
let thread_id = test_thread_id()?;
let tools = installed_tools_with_start(
runtime,
thread_id,
SessionSource::SubAgent(SubAgentSource::Review),
/*persistent_thread_state_available*/ true,
)
.await;
assert_eq!(Vec::<String>::new(), tool_names(&tools));
Ok(())
}
#[tokio::test]
async fn installed_goal_tools_reject_duplicate_goal_creation() -> anyhow::Result<()> {
let runtime = test_runtime().await?;
@@ -877,6 +911,21 @@ async fn thread_resume_rehydrates_active_goal_idle_accounting() -> anyhow::Resul
async fn installed_tools(
runtime: Arc<codex_state::StateRuntime>,
thread_id: ThreadId,
) -> Vec<Arc<dyn ToolExecutor<ToolCall>>> {
installed_tools_with_start(
runtime,
thread_id,
SessionSource::Cli,
/*persistent_thread_state_available*/ true,
)
.await
}
async fn installed_tools_with_start(
runtime: Arc<codex_state::StateRuntime>,
thread_id: ThreadId,
session_source: SessionSource,
persistent_thread_state_available: bool,
) -> Vec<Arc<dyn ToolExecutor<ToolCall>>> {
let mut builder = ExtensionRegistryBuilder::<()>::new();
install_with_backend(
@@ -889,13 +938,12 @@ async fn installed_tools(
let registry = builder.build();
let session_store = ExtensionData::new("session-1");
let thread_store = ExtensionData::new(thread_id.to_string());
let session_source = SessionSource::Cli;
for contributor in registry.thread_lifecycle_contributors() {
contributor
.on_thread_start(ThreadStartInput {
config: &(),
session_source: &session_source,
persistent_thread_state_available: true,
persistent_thread_state_available,
session_store: &session_store,
thread_store: &thread_store,
})
@@ -909,6 +957,10 @@ async fn installed_tools(
.collect()
}
fn tool_names(tools: &[Arc<dyn ToolExecutor<ToolCall>>]) -> Vec<String> {
tools.iter().map(|tool| tool.tool_name().name).collect()
}
struct GoalExtensionHarness {
registry: codex_extension_api::ExtensionRegistry<()>,
session_store: ExtensionData,
@@ -1070,6 +1122,7 @@ fn tool_call(tool_name: &str, call_id: &str, arguments: serde_json::Value) -> To
tool_name: codex_extension_api::ToolName::plain(tool_name),
truncation_policy: TruncationPolicy::Bytes(1024),
conversation_history: codex_extension_api::ConversationHistory::default(),
turn_item_emitter: Arc::new(NoopTurnItemEmitter),
payload: ToolPayload::Function {
arguments: arguments.to_string(),
},
+7
View File
@@ -4,6 +4,7 @@ use std::sync::Arc;
use codex_extension_api::ContextContributor;
use codex_extension_api::ExtensionData;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_extension_api::NoopTurnItemEmitter;
use codex_extension_api::PromptSlot;
use codex_extension_api::ToolCall;
use codex_extension_api::ToolContributor;
@@ -212,6 +213,7 @@ async fn add_ad_hoc_note_tool_creates_note_file() {
tool_name: memory_tool_name(crate::ADD_AD_HOC_NOTE_TOOL_NAME),
truncation_policy: TruncationPolicy::Bytes(1024),
conversation_history: codex_extension_api::ConversationHistory::default(),
turn_item_emitter: Arc::new(NoopTurnItemEmitter),
payload: payload.clone(),
})
.await
@@ -253,6 +255,7 @@ async fn add_ad_hoc_note_tool_rejects_paths_as_filenames() {
tool_name: memory_tool_name(crate::ADD_AD_HOC_NOTE_TOOL_NAME),
truncation_policy: TruncationPolicy::Bytes(1024),
conversation_history: codex_extension_api::ConversationHistory::default(),
turn_item_emitter: Arc::new(NoopTurnItemEmitter),
payload,
})
.await;
@@ -295,6 +298,7 @@ async fn read_tool_reads_memory_file() {
tool_name: memory_tool_name(crate::READ_TOOL_NAME),
truncation_policy: TruncationPolicy::Bytes(1024),
conversation_history: codex_extension_api::ConversationHistory::default(),
turn_item_emitter: Arc::new(NoopTurnItemEmitter),
payload: payload.clone(),
})
.await
@@ -340,6 +344,7 @@ async fn search_tool_accepts_multiple_queries() {
tool_name: memory_tool_name(crate::SEARCH_TOOL_NAME),
truncation_policy: TruncationPolicy::Bytes(1024),
conversation_history: codex_extension_api::ConversationHistory::default(),
turn_item_emitter: Arc::new(NoopTurnItemEmitter),
payload: payload.clone(),
})
.await
@@ -411,6 +416,7 @@ async fn search_tool_accepts_windowed_all_match_mode() {
tool_name: memory_tool_name(crate::SEARCH_TOOL_NAME),
truncation_policy: TruncationPolicy::Bytes(1024),
conversation_history: codex_extension_api::ConversationHistory::default(),
turn_item_emitter: Arc::new(NoopTurnItemEmitter),
payload: payload.clone(),
})
.await
@@ -462,6 +468,7 @@ async fn search_tool_rejects_legacy_single_query() {
tool_name: memory_tool_name(crate::SEARCH_TOOL_NAME),
truncation_policy: TruncationPolicy::Bytes(1024),
conversation_history: codex_extension_api::ConversationHistory::default(),
turn_item_emitter: Arc::new(NoopTurnItemEmitter),
payload,
})
.await;
+4
View File
@@ -62,7 +62,11 @@ pub use responses_api::mcp_tool_to_deferred_responses_api_tool;
pub use responses_api::mcp_tool_to_responses_api_tool;
pub use responses_api::tool_definition_to_responses_api_tool;
pub use tool_call::ConversationHistory;
pub use tool_call::ExtensionTurnItem;
pub use tool_call::NoopTurnItemEmitter;
pub use tool_call::ToolCall;
pub use tool_call::TurnItemEmissionFuture;
pub use tool_call::TurnItemEmitter;
pub use tool_config::ShellCommandBackendConfig;
pub use tool_config::ToolEnvironmentMode;
pub use tool_config::ToolUserShellType;
+57 -1
View File
@@ -1,8 +1,11 @@
use crate::FunctionCallError;
use crate::ToolName;
use crate::ToolPayload;
use codex_protocol::items::WebSearchItem;
use codex_protocol::models::ResponseItem;
use codex_utils_output_truncation::TruncationPolicy;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
/// Raw response history snapshot available when an extension tool is invoked.
@@ -23,17 +26,70 @@ impl ConversationHistory {
}
}
/// Future returned when an extension tool emits a visible turn-item lifecycle event.
pub type TurnItemEmissionFuture<'a> = Pin<Box<dyn Future<Output = ()> + Send + 'a>>;
/// Visible turn items that an extension fully owns and may emit as-is.
///
/// Add only item kinds that require no additional host finalization before
/// persistence or client delivery. Richer items need a host-owned publish path.
#[derive(Clone, Debug, PartialEq)]
pub enum ExtensionTurnItem {
WebSearch(WebSearchItem),
}
/// Host-provided capability for extension tools to emit finalized visible turn items.
///
/// Implementations route lifecycle events through the host's normal item event
/// pipeline, including any persistence and client delivery owned by the host.
pub trait TurnItemEmitter: Send + Sync {
/// Emits the beginning of one visible turn item.
fn emit_started<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a>;
/// Emits the completion of one visible turn item.
fn emit_completed<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a>;
}
/// Turn-item emitter used when a caller does not expose visible item emission.
#[derive(Debug, Default, Clone, Copy)]
pub struct NoopTurnItemEmitter;
impl TurnItemEmitter for NoopTurnItemEmitter {
fn emit_started<'a>(&'a self, _item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> {
Box::pin(std::future::ready(()))
}
fn emit_completed<'a>(&'a self, _item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> {
Box::pin(std::future::ready(()))
}
}
// TODO: this is temporary and will disappear in the next PR (as we make codex-extension-api generic on Invocation.
#[derive(Clone, Debug)]
#[derive(Clone)]
pub struct ToolCall {
pub turn_id: String,
pub call_id: String,
pub tool_name: ToolName,
pub truncation_policy: TruncationPolicy,
pub conversation_history: ConversationHistory,
pub turn_item_emitter: Arc<dyn TurnItemEmitter>,
pub payload: ToolPayload,
}
impl std::fmt::Debug for ToolCall {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ToolCall")
.field("turn_id", &self.turn_id)
.field("call_id", &self.call_id)
.field("tool_name", &self.tool_name)
.field("truncation_policy", &self.truncation_policy)
.field("conversation_history", &self.conversation_history)
.field("turn_item_emitter", &"<host turn item emitter>")
.field("payload", &self.payload)
.finish()
}
}
impl ToolCall {
pub fn function_arguments(&self) -> Result<&str, FunctionCallError> {
match &self.payload {
+1
View File
@@ -2300,6 +2300,7 @@ mod tests {
.into(),
active_permission_profile: None,
reasoning_effort: None,
initial_turns_page: None,
};
let started = started_thread_from_resume_response(
+18 -4
View File
@@ -2,6 +2,9 @@ set working-directory := "codex-rs"
set positional-arguments
rust_min_stack := "8388608" # 8 MiB
e2e_benchmark_packages := "codex-app-server-start-bench"
e2e_codex_bin := if os() == "windows" { "./target/release/codex.exe" } else { "./target/release/codex" }
e2e_smoke_codex_bin := if os() == "windows" { "./target/debug/codex.exe" } else { "./target/debug/codex" }
# Display help
help:
@@ -30,6 +33,11 @@ app-server-test-client *args:
cargo build -p codex-cli
cargo run -p codex-app-server-test-client -- --codex-bin ./target/debug/codex "$@"
# Run end-to-end performance benchmarks that require a built Codex binary.
bench-e2e *args:
cargo build --release -p codex-cli --bin codex
for package in {{ e2e_benchmark_packages }}; do CODEX_BIN="{{ e2e_codex_bin }}" cargo run --release -p "$package" -- --bench "$@"; done
# Format Rust and Python SDK code.
fmt:
cargo fmt -- --config imports_granularity=Item 2>/dev/null
@@ -55,13 +63,19 @@ test *args:
RUST_MIN_STACK={{ rust_min_stack }} cargo nextest run --no-fail-fast "$@"
just bench-smoke
# Run explicit workspace benchmark targets.
bench *args:
# Run unit-test-style benchmark targets managed entirely by Cargo.
bench-unit *args:
cargo bench --workspace --bench '*' "$@"
# Run benchmark targets once to ensure they start successfully.
# Smoke Cargo-managed benchmarks and compile e2e runners without measured binaries.
bench-smoke:
just bench -- --test
just bench-unit -- --test
for package in {{ e2e_benchmark_packages }}; do cargo build -p "$package" --bin "$package"; done
# Run end-to-end performance benchmark targets once.
bench-e2e-smoke:
cargo build -p codex-cli --bin codex
for package in {{ e2e_benchmark_packages }}; do CODEX_BIN="{{ e2e_smoke_codex_bin }}" cargo run -p "$package" -- --test; done
# Build and run Codex from source using Bazel.
# Note we have to use the combination of `[no-cd]` and `--run_under="cd $PWD &&"`