openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-03 19:53:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add thread metadata

Browse Source
This commit is contained in:
Guinness Chen
2026-03-24 22:14:27 -07:00
parent 8c62829a2b
commit 4573855b35
43 changed files with 934 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
+28
View File
@@ -2554,6 +2554,15 @@
"ephemeral": {
"type": "boolean"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"type": [
"object",
"null"
]
},
"model": {
"description": "Configuration overrides for the forked thread, if any.",
"type": [
@@ -2736,6 +2745,16 @@
],
"description": "Patch the stored Git metadata for this thread. Omit a field to leave it unchanged, set it to `null` to clear it, or provide a string to replace the stored value."
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"description": "Replace the stored client-defined metadata for this thread.",
"type": [
"object",
"null"
]
},
"threadId": {
"type": "string"
}
@@ -3032,6 +3051,15 @@
"null"
]
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"type": [
"object",
"null"
]
},
"model": {
"type": [
"string",
codex-rs/app-server-protocol/schema/json/ServerNotification.json
+8
View File
@@ -2244,6 +2244,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
+36
View File
@@ -12023,6 +12023,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
@@ -12211,6 +12219,15 @@
"ephemeral": {
"type": "boolean"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"type": [
"object",
"null"
]
},
"model": {
"description": "Configuration overrides for the forked thread, if any.",
"type": [
@@ -13151,6 +13168,16 @@
],
"description": "Patch the stored Git metadata for this thread. Omit a field to leave it unchanged, set it to `null` to clear it, or provide a string to replace the stored value."
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"description": "Replace the stored client-defined metadata for this thread.",
"type": [
"object",
"null"
]
},
"threadId": {
"type": "string"
}
@@ -13712,6 +13739,15 @@
"null"
]
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"type": [
"object",
"null"
]
},
"model": {
"type": [
"string",
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
+36
View File
@@ -9770,6 +9770,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
@@ -9958,6 +9966,15 @@
"ephemeral": {
"type": "boolean"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"type": [
"object",
"null"
]
},
"model": {
"description": "Configuration overrides for the forked thread, if any.",
"type": [
@@ -10898,6 +10915,16 @@
],
"description": "Patch the stored Git metadata for this thread. Omit a field to leave it unchanged, set it to `null` to clear it, or provide a string to replace the stored value."
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"description": "Replace the stored client-defined metadata for this thread.",
"type": [
"object",
"null"
]
},
"threadId": {
"type": "string"
}
@@ -11459,6 +11486,15 @@
"null"
]
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"type": [
"object",
"null"
]
},
"model": {
"type": [
"string",
codex-rs/app-server-protocol/schema/json/v2/ThreadForkParams.json
+9
View File
@@ -126,6 +126,15 @@
"ephemeral": {
"type": "boolean"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"type": [
"object",
"null"
]
},
"model": {
"description": "Configuration overrides for the forked thread, if any.",
"type": [
codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json
+8
View File
@@ -1056,6 +1056,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json
+8
View File
@@ -814,6 +814,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateParams.json
+10
View File
@@ -40,6 +40,16 @@
],
"description": "Patch the stored Git metadata for this thread. Omit a field to leave it unchanged, set it to `null` to clear it, or provide a string to replace the stored value."
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"description": "Replace the stored client-defined metadata for this thread.",
"type": [
"object",
"null"
]
},
"threadId": {
"type": "string"
}
codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json
+8
View File
@@ -814,6 +814,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json
+8
View File
@@ -814,6 +814,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json
+8
View File
@@ -1056,6 +1056,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json
+8
View File
@@ -814,6 +814,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json
+9
View File
@@ -156,6 +156,15 @@
"null"
]
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"type": [
"object",
"null"
]
},
"model": {
"type": [
"string",
codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json
+8
View File
@@ -1056,6 +1056,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json
+8
View File
@@ -814,6 +814,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json
+8
View File
@@ -814,6 +814,14 @@
"id": {
"type": "string"
},
"metadata": {
"additionalProperties": {
"type": "string"
},
"default": {},
"description": "Arbitrary client-defined metadata.",
"type": "object"
},
"modelProvider": {
"description": "Model provider used for this thread (for example, 'openai').",
"type": "string"
codex-rs/app-server-protocol/schema/typescript/v2/Thread.ts
+4
View File
@@ -63,6 +63,10 @@ gitInfo: GitInfo | null,
* Optional user-facing thread title.
*/
name: string | null,
/**
* Arbitrary client-defined metadata.
*/
metadata: { [key in string]?: string },
/**
* Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read`
* (when `includeTurns` is true) responses.
codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts
+1 -1
View File
@@ -27,7 +27,7 @@ model?: string | null, modelProvider?: string | null, serviceTier?: ServiceTier
* Override where approval requests are routed for review on this thread
* and subsequent turns.
*/
approvalsReviewer?: ApprovalsReviewer | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, baseInstructions?: string | null, developerInstructions?: string | null, ephemeral?: boolean, /**
approvalsReviewer?: ApprovalsReviewer | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, baseInstructions?: string | null, developerInstructions?: string | null, metadata?: { [key in string]?: string } | null, ephemeral?: boolean, /**
* If true, persist additional rollout EventMsg variants required to
* reconstruct a richer thread history on subsequent resume/fork/read.
*/
codex-rs/app-server-protocol/schema/typescript/v2/ThreadMetadataUpdateParams.ts
+4
View File
@@ -4,6 +4,10 @@
import type { ThreadMetadataGitInfoUpdateParams } from "./ThreadMetadataGitInfoUpdateParams";
export type ThreadMetadataUpdateParams = { threadId: string,
/**
* Replace the stored client-defined metadata for this thread.
*/
metadata?: { [key in string]?: string } | null,
/**
* Patch the stored Git metadata for this thread.
* Omit a field to leave it unchanged, set it to `null` to clear it, or
codex-rs/app-server-protocol/schema/typescript/v2/ThreadStartParams.ts
+1 -1
View File
@@ -12,7 +12,7 @@ export type ThreadStartParams = {model?: string | null, modelProvider?: string |
* Override where approval requests are routed for review on this thread
* and subsequent turns.
*/
approvalsReviewer?: ApprovalsReviewer | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, serviceName?: string | null, baseInstructions?: string | null, developerInstructions?: string | null, personality?: Personality | null, ephemeral?: boolean | null, /**
approvalsReviewer?: ApprovalsReviewer | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, serviceName?: string | null, baseInstructions?: string | null, developerInstructions?: string | null, personality?: Personality | null, metadata?: { [key in string]?: string } | null, ephemeral?: boolean | null, /**
* If true, opt into emitting raw Responses API items on the event stream.
* This is for internal use only (e.g. Codex Cloud).
*/
codex-rs/app-server-protocol/src/protocol/v2.rs
+10
View File
@@ -2586,6 +2586,8 @@ pub struct ThreadStartParams {
#[ts(optional = nullable)]
pub personality: Option<Personality>,
#[ts(optional = nullable)]
pub metadata: Option<BTreeMap<String, String>>,
#[ts(optional = nullable)]
pub ephemeral: Option<bool>,
#[experimental("thread/start.dynamicTools")]
#[ts(optional = nullable)]
@@ -2778,6 +2780,8 @@ pub struct ThreadForkParams {
pub base_instructions: Option<String>,
#[ts(optional = nullable)]
pub developer_instructions: Option<String>,
#[ts(optional = nullable)]
pub metadata: Option<BTreeMap<String, String>>,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub ephemeral: bool,
/// If true, persist additional rollout EventMsg variants required to
@@ -2904,6 +2908,9 @@ pub struct ThreadSetNameResponse {}
#[ts(export_to = "v2/")]
pub struct ThreadMetadataUpdateParams {
pub thread_id: String,
/// Replace the stored client-defined metadata for this thread.
#[ts(optional = nullable)]
pub metadata: Option<BTreeMap<String, String>>,
/// Patch the stored Git metadata for this thread.
/// Omit a field to leave it unchanged, set it to `null` to clear it, or
/// provide a string to replace the stored value.
@@ -3620,6 +3627,9 @@ pub struct Thread {
pub git_info: Option<GitInfo>,
/// Optional user-facing thread title.
pub name: Option<String>,
/// Arbitrary client-defined metadata.
#[serde(default)]
pub metadata: BTreeMap<String, String>,
/// Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read`
/// (when `includeTurns` is true) responses.
/// For all other responses and notifications returning a Thread,
codex-rs/app-server/README.md
+32 -13
View File
@@ -123,13 +123,13 @@ Example with notification opt-out:
## API Overview
- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread.
- `thread/start` — create a new thread; accepts optional client-defined `metadata` (`string -> string`, up to 16 entries with keys/values <= 512 chars), emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for that thread.
- `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it.
- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread.
- `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. Accepts optional client-defined `metadata` plus `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread.
- `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded, plus any stored client `metadata`.
- `thread/loaded/list` — list the thread ids currently loaded in memory.
- `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
- `thread/metadata/update` — patch stored thread metadata in sqlite; currently supports updating persisted `gitInfo` fields and returns the refreshed `thread`.
- `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded, plus any stored client `metadata`.
- `thread/metadata/update` — replace stored client `metadata` and/or patch stored `gitInfo` fields in sqlite, then return the refreshed `thread`.
- `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`).
- `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success and emits `thread/archived`.
- `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server shuts down and unloads the thread, then emits `thread/closed`.
@@ -201,6 +201,7 @@ Start a fresh thread when you need a new Codex conversation.
"sandbox": "workspaceWrite",
"personality": "friendly",
"serviceName": "my_app_server_client", // optional metrics tag (`service_name`)
"metadata": { "client": "vscode", "surface": "chat" },
// Experimental: requires opt-in
"dynamicTools": [
{
@@ -222,7 +223,8 @@ Start a fresh thread when you need a new Codex conversation.
"id": "thr_123",
"preview": "",
"modelProvider": "openai",
"createdAt": 1730910000
"createdAt": 1730910000,
"metadata": { "client": "vscode", "surface": "chat" }
}
} }
{ "method": "thread/started", "params": { "thread": { … } } }
@@ -244,10 +246,14 @@ Example:
{ "id": 11, "result": { "thread": { "id": "thr_123", … } } }
```
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. 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:
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. If the source thread is actively running, the fork snapshots it as if the current turn had been interrupted first. You can also set replacement client `metadata` on the fork. Pass `ephemeral: true` when the fork should stay in-memory only:
```json
{ "method": "thread/fork", "id": 12, "params": { "threadId": "thr_123", "ephemeral": true } }
{ "method": "thread/fork", "id": 12, "params": {
"threadId": "thr_123",
"metadata": { "origin": "review" },
"ephemeral": true
} }
{ "id": 12, "result": { "thread": { "id": "thr_456", … } } }
{ "method": "thread/started", "params": { "thread": { … } } }
```
@@ -338,7 +344,7 @@ If this was the last subscriber, the server unloads the thread and emits `thread
### Example: Read a thread
Use `thread/read` to fetch a stored thread by id without resuming it. Pass `includeTurns` when you want the rollout history loaded into `thread.turns`. The returned thread includes `agentNickname` and `agentRole` for AgentControl-spawned thread sub-agents when available.
Use `thread/read` to fetch a stored thread by id without resuming it. Pass `includeTurns` when you want the rollout history loaded into `thread.turns`. The returned thread includes `agentNickname` and `agentRole` for AgentControl-spawned thread sub-agents when available, plus stored client `metadata` when present.
```json
{ "method": "thread/read", "id": 22, "params": { "threadId": "thr_123" } }
@@ -356,25 +362,38 @@ Use `thread/read` to fetch a stored thread by id without resuming it. Pass `incl
### Example: Update stored thread metadata
Use `thread/metadata/update` to patch sqlite-backed metadata for a thread without resuming it. Today this supports persisted `gitInfo`; omitted fields are left unchanged, while explicit `null` clears a stored value.
Use `thread/metadata/update` to update sqlite-backed metadata for a thread without resuming it. `metadata` replaces the stored client-defined key/value map. `gitInfo` remains patch-style: omitted fields are left unchanged, while explicit `null` clears a stored value.
```json
{ "method": "thread/metadata/update", "id": 24, "params": {
"threadId": "thr_123",
"gitInfo": { "branch": "feature/sidebar-pr" }
"metadata": { "client": "desktop", "surface": "composer" }
} }
{ "id": 24, "result": {
"thread": {
"id": "thr_123",
"metadata": { "client": "desktop", "surface": "composer" }
}
} }
```
```json
{ "method": "thread/metadata/update", "id": 25, "params": {
"threadId": "thr_123",
"gitInfo": { "branch": "feature/sidebar-pr" }
} }
{ "id": 25, "result": {
"thread": {
"id": "thr_123",
"gitInfo": { "sha": null, "branch": "feature/sidebar-pr", "originUrl": null }
}
} }
{ "method": "thread/metadata/update", "id": 25, "params": {
{ "method": "thread/metadata/update", "id": 26, "params": {
"threadId": "thr_123",
"gitInfo": { "branch": null }
} }
{ "id": 25, "result": {
{ "id": 26, "result": {
"thread": {
"id": "thr_123",
"gitInfo": null
codex-rs/app-server/src/codex_message_processor.rs
+305 -67
View File
@@ -1892,6 +1892,7 @@ impl CodexMessageProcessor {
service_name,
base_instructions,
developer_instructions,
metadata,
dynamic_tools,
mock_experimental_field: _mock_experimental_field,
experimental_raw_events,
@@ -1899,6 +1900,13 @@ impl CodexMessageProcessor {
ephemeral,
persist_extended_history,
} = params;
let thread_metadata = match metadata.map(validate_thread_metadata).transpose() {
Ok(metadata) => metadata.unwrap_or_default(),
Err(message) => {
self.send_invalid_request_error(request_id, message).await;
return;
}
};
let mut typesafe_overrides = self.build_thread_config_overrides(
model,
model_provider,
@@ -1936,6 +1944,7 @@ impl CodexMessageProcessor {
dynamic_tools,
persist_extended_history,
service_name,
thread_metadata,
experimental_raw_events,
request_trace,
)
@@ -2009,6 +2018,7 @@ impl CodexMessageProcessor {
dynamic_tools: Option<Vec<ApiDynamicToolSpec>>,
persist_extended_history: bool,
service_name: Option<String>,
thread_metadata: BTreeMap<String, String>,
experimental_raw_events: bool,
request_trace: Option<W3cTraceContext>,
) {
@@ -2081,11 +2091,11 @@ impl CodexMessageProcessor {
Ok(new_conv) => {
let NewThread {
thread_id,
thread,
thread: started_thread,
session_configured,
..
} = new_conv;
let config_snapshot = thread
let config_snapshot = started_thread
.config_snapshot()
.instrument(tracing::info_span!(
"app_server.thread_start.config_snapshot",
@@ -2097,6 +2107,32 @@ impl CodexMessageProcessor {
&config_snapshot,
session_configured.rollout_path.clone(),
);
thread.metadata = thread_metadata.clone();
if let (Some(state_db_ctx), Some(rollout_path)) = (
started_thread.state_db(),
session_configured.rollout_path.as_ref(),
) && let Err(message) = persist_thread_metadata_from_snapshot(
state_db_ctx.as_ref(),
thread_id,
rollout_path.as_path(),
&config_snapshot,
&thread_metadata,
)
.await
{
listener_task_context
.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message,
data: None,
},
)
.await;
return;
}
// Auto-attach a thread listener when starting a thread.
Self::log_listener_attach_result(
@@ -2434,8 +2470,16 @@ impl CodexMessageProcessor {
) {
let ThreadMetadataUpdateParams {
thread_id,
metadata,
git_info,
} = params;
let metadata = match metadata.map(validate_thread_metadata).transpose() {
Ok(metadata) => metadata,
Err(message) => {
self.send_invalid_request_error(request_id, message).await;
return;
}
};
let thread_uuid = match ThreadId::from_string(&thread_id) {
Ok(id) => id,
@@ -2446,21 +2490,26 @@ impl CodexMessageProcessor {
}
};
let Some(ThreadMetadataGitInfoUpdateParams {
sha,
branch,
origin_url,
}) = git_info
else {
if metadata.is_none() && git_info.is_none() {
self.send_invalid_request_error(
request_id,
"gitInfo must include at least one field".to_string(),
"metadata or gitInfo must include at least one field".to_string(),
)
.await;
return;
};
}
if sha.is_none() && branch.is_none() && origin_url.is_none() {
let (sha, branch, origin_url) = git_info
.map(
|ThreadMetadataGitInfoUpdateParams {
sha,
branch,
origin_url,
}| (sha, branch, origin_url),
)
.unwrap_or((None, None, None));
if metadata.is_none() && sha.is_none() && branch.is_none() && origin_url.is_none() {
self.send_invalid_request_error(
request_id,
"gitInfo must include at least one field".to_string(),
@@ -2540,36 +2589,77 @@ impl CodexMessageProcessor {
None => None,
};
let updated = match state_db_ctx
.update_thread_git_info(
thread_uuid,
git_sha.as_ref().map(|value| value.as_deref()),
git_branch.as_ref().map(|value| value.as_deref()),
git_origin_url.as_ref().map(|value| value.as_deref()),
)
.await
{
Ok(updated) => updated,
Err(err) => {
if let Some(metadata) = metadata.as_ref() {
let metadata_json = match serde_json::to_string(metadata) {
Ok(metadata_json) => metadata_json,
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to serialize thread metadata for {thread_uuid}: {err}"),
)
.await;
return;
}
};
let updated = match state_db_ctx
.update_thread_metadata_json(thread_uuid, metadata_json.as_str())
.await
{
Ok(updated) => updated,
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to update thread metadata for {thread_uuid}: {err}"),
)
.await;
return;
}
};
if !updated {
self.send_internal_error(
request_id,
format!("failed to update thread metadata for {thread_uuid}: {err}"),
format!("thread metadata disappeared before update completed: {thread_uuid}"),
)
.await;
return;
}
};
if !updated {
self.send_internal_error(
request_id,
format!("thread metadata disappeared before update completed: {thread_uuid}"),
)
.await;
return;
}
let Some(summary) =
read_summary_from_state_db_context_by_thread_id(Some(&state_db_ctx), thread_uuid).await
if git_sha.is_some() || git_branch.is_some() || git_origin_url.is_some() {
let updated = match state_db_ctx
.update_thread_git_info(
thread_uuid,
git_sha.as_ref().map(|value| value.as_deref()),
git_branch.as_ref().map(|value| value.as_deref()),
git_origin_url.as_ref().map(|value| value.as_deref()),
)
.await
{
Ok(updated) => updated,
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to update thread metadata for {thread_uuid}: {err}"),
)
.await;
return;
}
};
if !updated {
self.send_internal_error(
request_id,
format!("thread metadata disappeared before update completed: {thread_uuid}"),
)
.await;
return;
}
}
let Some(persisted_metadata) = read_thread_metadata_from_state_db_context_by_thread_id(
Some(&state_db_ctx),
thread_uuid,
)
.await
else {
self.send_internal_error(
request_id,
@@ -2579,7 +2669,8 @@ impl CodexMessageProcessor {
return;
};
let mut thread = summary_to_thread(summary);
let mut thread = summary_to_thread(summary_from_thread_metadata(&persisted_metadata));
merge_persisted_thread_metadata(&mut thread, &persisted_metadata);
self.attach_thread_name(thread_uuid, &mut thread).await;
thread.status = resolve_thread_status(
self.thread_watch_manager
@@ -3140,12 +3231,22 @@ impl CodexMessageProcessor {
let mut threads = Vec::with_capacity(summaries.len());
let mut thread_ids = HashSet::with_capacity(summaries.len());
let mut status_ids = Vec::with_capacity(summaries.len());
let state_db_ctx = get_state_db(&self.config).await;
for summary in summaries {
let conversation_id = summary.conversation_id;
thread_ids.insert(conversation_id);
let thread = summary_to_thread(summary);
let mut thread = summary_to_thread(summary);
if let Some(persisted_metadata) =
read_thread_metadata_from_state_db_context_by_thread_id(
state_db_ctx.as_ref(),
conversation_id,
)
.await
{
merge_persisted_thread_metadata(&mut thread, &persisted_metadata);
}
status_ids.push(thread.id.clone());
threads.push((conversation_id, thread));
}
@@ -3253,12 +3354,15 @@ impl CodexMessageProcessor {
let loaded_thread = self.thread_manager.get_thread(thread_uuid).await.ok();
let loaded_thread_state_db = loaded_thread.as_ref().and_then(|thread| thread.state_db());
let db_summary = if let Some(state_db_ctx) = loaded_thread_state_db.as_ref() {
read_summary_from_state_db_context_by_thread_id(Some(state_db_ctx), thread_uuid).await
let db_metadata = if let Some(state_db_ctx) = loaded_thread_state_db.as_ref() {
read_thread_metadata_from_state_db_context_by_thread_id(Some(state_db_ctx), thread_uuid)
.await
} else {
read_summary_from_state_db_by_thread_id(&self.config, thread_uuid).await
read_thread_metadata_from_state_db_by_thread_id(&self.config, thread_uuid).await
};
let mut rollout_path = db_summary.as_ref().map(|summary| summary.path.clone());
let mut rollout_path = db_metadata
.as_ref()
.map(|metadata| metadata.rollout_path.clone());
if rollout_path.is_none() || include_turns {
rollout_path =
match find_thread_path_by_id_str(&self.config.codex_home, &thread_uuid.to_string())
@@ -3283,17 +3387,31 @@ impl CodexMessageProcessor {
};
}
if include_turns && rollout_path.is_none() && db_summary.is_some() {
self.send_internal_error(
request_id,
format!("failed to locate rollout for thread {thread_uuid}"),
)
.await;
return;
if include_turns && rollout_path.is_none() {
if loaded_thread.is_some() {
self.send_invalid_request_error(
request_id,
format!(
"thread {thread_uuid} is not materialized yet; includeTurns is unavailable before first user message"
),
)
.await;
return;
}
if db_metadata.is_some() {
self.send_internal_error(
request_id,
format!("failed to locate rollout for thread {thread_uuid}"),
)
.await;
return;
}
}
let mut thread = if let Some(summary) = db_summary {
summary_to_thread(summary)
let mut thread = if let Some(persisted_metadata) = db_metadata.as_ref() {
let mut thread = summary_to_thread(summary_from_thread_metadata(persisted_metadata));
merge_persisted_thread_metadata(&mut thread, persisted_metadata);
thread
} else if let Some(rollout_path) = rollout_path.as_ref() {
let fallback_provider = self.config.model_provider_id.as_str();
match read_summary_from_rollout(rollout_path, fallback_provider).await {
@@ -3980,9 +4098,17 @@ impl CodexMessageProcessor {
config: cli_overrides,
base_instructions,
developer_instructions,
metadata,
ephemeral,
persist_extended_history,
} = params;
let thread_metadata = match metadata.map(validate_thread_metadata).transpose() {
Ok(metadata) => metadata.unwrap_or_default(),
Err(message) => {
self.send_invalid_request_error(request_id, message).await;
return;
}
};
let (rollout_path, source_thread_id) = if let Some(path) = path {
(path, None)
@@ -4130,6 +4256,7 @@ impl CodexMessageProcessor {
return;
}
};
let fork_config_snapshot = forked_thread.config_snapshot().await;
// Auto-attach a conversation listener when forking a thread.
Self::log_listener_attach_result(
@@ -4168,10 +4295,9 @@ impl CodexMessageProcessor {
}
}
} else {
let config_snapshot = forked_thread.config_snapshot().await;
// forked thread names do not inherit the source thread name
let mut thread =
build_thread_from_snapshot(thread_id, &config_snapshot, /*path*/ None);
build_thread_from_snapshot(thread_id, &fork_config_snapshot, /*path*/ None);
let history_items = match read_rollout_items_from_rollout(rollout_path.as_path()).await
{
Ok(items) => items,
@@ -4200,6 +4326,7 @@ impl CodexMessageProcessor {
}
thread
};
thread.metadata = thread_metadata.clone();
if let Some(fork_rollout_path) = session_configured.rollout_path.as_ref()
&& let Err(message) = populate_thread_turns(
@@ -4212,6 +4339,20 @@ impl CodexMessageProcessor {
self.send_internal_error(request_id, message).await;
return;
}
if let Some(fork_rollout_path) = session_configured.rollout_path.as_ref()
&& let Some(state_db_ctx) = forked_thread.state_db()
&& let Err(message) = persist_thread_metadata_from_snapshot(
state_db_ctx.as_ref(),
thread_id,
fork_rollout_path.as_path(),
&fork_config_snapshot,
&thread_metadata,
)
.await
{
self.send_internal_error(request_id, message).await;
return;
}
self.thread_watch_manager
.upsert_thread_silently(thread.clone())
@@ -7983,25 +8124,42 @@ async fn read_history_cwd_from_state_db(
}
}
async fn read_thread_metadata_from_state_db_by_thread_id(
config: &Config,
thread_id: ThreadId,
) -> Option<ThreadMetadata> {
let state_db_ctx = get_state_db(config).await;
read_thread_metadata_from_state_db_context_by_thread_id(state_db_ctx.as_ref(), thread_id).await
}
async fn read_thread_metadata_from_state_db_context_by_thread_id(
state_db_ctx: Option<&StateDbHandle>,
thread_id: ThreadId,
) -> Option<ThreadMetadata> {
let state_db_ctx = state_db_ctx?;
match state_db_ctx.get_thread(thread_id).await {
Ok(Some(metadata)) => Some(metadata),
Ok(None) | Err(_) => None,
}
}
async fn read_summary_from_state_db_by_thread_id(
config: &Config,
thread_id: ThreadId,
) -> Option<ConversationSummary> {
let state_db_ctx = get_state_db(config).await;
read_summary_from_state_db_context_by_thread_id(state_db_ctx.as_ref(), thread_id).await
read_thread_metadata_from_state_db_by_thread_id(config, thread_id)
.await
.map(|metadata| summary_from_thread_metadata(&metadata))
}
async fn read_summary_from_state_db_context_by_thread_id(
state_db_ctx: Option<&StateDbHandle>,
thread_id: ThreadId,
) -> Option<ConversationSummary> {
let state_db_ctx = state_db_ctx?;
let metadata = match state_db_ctx.get_thread(thread_id).await {
Ok(Some(metadata)) => metadata,
Ok(None) | Err(_) => return None,
};
Some(summary_from_thread_metadata(&metadata))
read_thread_metadata_from_state_db_context_by_thread_id(state_db_ctx, thread_id)
.await
.map(|metadata| summary_from_thread_metadata(&metadata))
}
async fn summary_from_thread_list_item(
@@ -8294,18 +8452,29 @@ async fn load_thread_summary_for_rollout(
)
})?;
if let Some(persisted_metadata) = persisted_metadata {
merge_mutable_thread_metadata(
&mut thread,
summary_to_thread(summary_from_thread_metadata(persisted_metadata)),
);
} else if let Some(summary) = read_summary_from_state_db_by_thread_id(config, thread_id).await {
merge_mutable_thread_metadata(&mut thread, summary_to_thread(summary));
merge_persisted_thread_metadata(&mut thread, persisted_metadata);
} else if let Some(persisted_metadata) =
read_thread_metadata_from_state_db_by_thread_id(config, thread_id).await
{
merge_persisted_thread_metadata(&mut thread, &persisted_metadata);
}
Ok(thread)
}
fn merge_mutable_thread_metadata(thread: &mut Thread, persisted_thread: Thread) {
thread.git_info = persisted_thread.git_info;
fn merge_persisted_thread_metadata(thread: &mut Thread, persisted_metadata: &ThreadMetadata) {
thread.git_info = if persisted_metadata.git_sha.is_none()
&& persisted_metadata.git_branch.is_none()
&& persisted_metadata.git_origin_url.is_none()
{
None
} else {
Some(ApiGitInfo {
sha: persisted_metadata.git_sha.clone(),
branch: persisted_metadata.git_branch.clone(),
origin_url: persisted_metadata.git_origin_url.clone(),
})
};
thread.metadata = parse_thread_metadata_json(persisted_metadata.metadata_json.as_str());
}
fn preview_from_rollout_items(items: &[RolloutItem]) -> String {
@@ -8398,6 +8567,7 @@ fn build_thread_from_snapshot(
source: config_snapshot.session_source.clone().into(),
git_info: None,
name: None,
metadata: BTreeMap::new(),
turns: Vec::new(),
}
}
@@ -8440,10 +8610,78 @@ pub(crate) fn summary_to_thread(summary: ConversationSummary) -> Thread {
source: source.into(),
git_info,
name: None,
metadata: BTreeMap::new(),
turns: Vec::new(),
}
}
fn parse_thread_metadata_json(metadata_json: &str) -> BTreeMap<String, String> {
serde_json::from_str(metadata_json).unwrap_or_default()
}
async fn persist_thread_metadata_from_snapshot(
state_db_ctx: &StateRuntime,
thread_id: ThreadId,
rollout_path: &Path,
config_snapshot: &ThreadConfigSnapshot,
thread_metadata: &BTreeMap<String, String>,
) -> Result<(), String> {
let metadata_json = serde_json::to_string(thread_metadata)
.map_err(|err| format!("failed to serialize thread metadata for {thread_id}: {err}"))?;
let mut builder = ThreadMetadataBuilder::new(
thread_id,
rollout_path.to_path_buf(),
Utc::now(),
config_snapshot.session_source.clone(),
);
builder.model_provider = Some(config_snapshot.model_provider_id.clone());
builder.cwd = config_snapshot.cwd.clone();
builder.cli_version = Some(env!("CARGO_PKG_VERSION").to_string());
builder.sandbox_policy = config_snapshot.sandbox_policy.clone();
builder.approval_mode = config_snapshot.approval_policy;
builder.metadata_json = metadata_json.clone();
let metadata = builder.build(config_snapshot.model_provider_id.as_str());
let inserted = state_db_ctx
.insert_thread_if_absent(&metadata)
.await
.map_err(|err| format!("failed to create thread metadata for {thread_id}: {err}"))?;
if !inserted {
let updated = state_db_ctx
.update_thread_metadata_json(thread_id, metadata_json.as_str())
.await
.map_err(|err| format!("failed to persist thread metadata for {thread_id}: {err}"))?;
if !updated {
return Err(format!(
"thread metadata disappeared before metadata persisted: {thread_id}"
));
}
}
Ok(())
}
fn validate_thread_metadata(
metadata: BTreeMap<String, String>,
) -> std::result::Result<BTreeMap<String, String>, String> {
if metadata.len() > 16 {
return Err("metadata must include at most 16 entries".to_string());
}
for (key, value) in &metadata {
if key.chars().count() > 512 {
return Err(format!("metadata key exceeds 512 characters: {key}"));
}
if value.chars().count() > 512 {
return Err(format!(
"metadata value exceeds 512 characters for key: {key}"
));
}
}
Ok(metadata)
}
#[cfg(test)]
mod tests {
use super::*;
codex-rs/app-server/src/thread_status.rs
+1
View File
@@ -802,6 +802,7 @@ mod tests {
source,
git_info: None,
name: None,
metadata: std::collections::BTreeMap::new(),
turns: Vec::new(),
}
}
codex-rs/app-server/tests/suite/v2/skills_list.rs
+1
View File
@@ -240,6 +240,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<(
base_instructions: None,
developer_instructions: None,
personality: None,
metadata: None,
ephemeral: None,
dynamic_tools: None,
mock_experimental_field: None,
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+15
View File
@@ -29,6 +29,7 @@ use codex_core::auth::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::collections::BTreeMap;
use std::path::Path;
use tempfile::TempDir;
use tokio::time::timeout;
@@ -75,9 +76,12 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let metadata = BTreeMap::from([("origin".to_string(), "sidebar".to_string())]);
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
metadata: Some(metadata.clone()),
..Default::default()
})
.await?;
@@ -99,6 +103,11 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
Some(&Value::Null),
"forked threads do not inherit a name; expected `name: null`"
);
assert_eq!(
thread_json.get("metadata"),
Some(&json!({ "origin": "sidebar" })),
"forked threads should serialize client metadata"
);
let after_contents = std::fs::read_to_string(&original_path)?;
assert_eq!(
@@ -110,6 +119,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
assert_eq!(thread.preview, preview);
assert_eq!(thread.model_provider, "mock_provider");
assert_eq!(thread.status, ThreadStatus::Idle);
assert_eq!(thread.metadata, metadata);
let thread_path = thread.path.clone().expect("thread path");
assert!(thread_path.is_absolute());
assert_ne!(thread_path, original_path);
@@ -170,6 +180,11 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
Some(&Value::Null),
"thread/started must serialize `name: null` when unset"
);
assert_eq!(
started_thread_json.get("metadata"),
Some(&json!({ "origin": "sidebar" })),
"thread/started must serialize client metadata for forks"
);
let started: ThreadStartedNotification =
serde_json::from_value(notif.params.expect("params must be present"))?;
assert_eq!(started.thread, thread);
codex-rs/app-server/tests/suite/v2/thread_list.rs
+100
View File
@@ -3,6 +3,7 @@ use app_test_support::McpProcess;
use app_test_support::create_fake_rollout;
use app_test_support::create_fake_rollout_with_source;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::create_mock_responses_server_sequence;
use app_test_support::rollout_path;
use app_test_support::to_response;
@@ -33,6 +34,7 @@ use codex_protocol::protocol::SubAgentSource;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use std::cmp::Reverse;
use std::collections::BTreeMap;
use std::fs;
use std::fs::FileTimes;
use std::fs::OpenOptions;
@@ -311,6 +313,36 @@ stream_max_retries = 0
)
}
fn create_runtime_sqlite_config(
codex_home: &std::path::Path,
server_uri: &str,
) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
format!(
r#"
model = "mock-model"
approval_policy = "never"
sandbox_mode = "read-only"
suppress_unstable_features_warning = true
model_provider = "mock_provider"
[features]
sqlite = true
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
"#
),
)
}
#[tokio::test]
async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> {
let codex_home = TempDir::new()?;
@@ -591,6 +623,74 @@ sqlite = true
Ok(())
}
#[tokio::test]
async fn thread_list_returns_persisted_metadata() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_runtime_sqlite_config(codex_home.path(), &server.uri())?;
let state_db =
codex_state::StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into())
.await?;
state_db.mark_backfill_complete(None).await?;
let mut mcp = init_mcp(codex_home.path()).await?;
let metadata = BTreeMap::from([("surface".to_string(), "picker".to_string())]);
let start_id = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
metadata: Some(metadata.clone()),
..Default::default()
})
.await?;
let start_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let turn_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "materialize thread".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_id)),
)
.await??;
let _: TurnStartResponse = to_response::<TurnStartResponse>(turn_resp)?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let ThreadListResponse { data, .. } = list_threads(
&mut mcp,
None,
Some(10),
Some(vec!["mock_provider".to_string()]),
None,
None,
)
.await?;
let listed = data
.into_iter()
.find(|listed_thread| listed_thread.id == thread.id)
.expect("thread/list should include the started thread");
assert_eq!(listed.metadata, metadata);
assert_eq!(listed.status, ThreadStatus::Idle);
Ok(())
}
#[tokio::test]
async fn thread_list_empty_source_kinds_defaults_to_interactive_only() -> Result<()> {
let codex_home = TempDir::new()?;
codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs
+76
View File
@@ -26,6 +26,7 @@ use codex_protocol::protocol::GitInfo as RolloutGitInfo;
use codex_state::StateRuntime;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::collections::BTreeMap;
use std::fs;
use std::path::Path;
use std::sync::Arc;
@@ -59,6 +60,7 @@ async fn thread_metadata_update_patches_git_branch_and_returns_updated_thread()
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread.id.clone(),
metadata: None,
git_info: Some(ThreadMetadataGitInfoUpdateParams {
sha: None,
branch: Some(Some("feature/sidebar-pr".to_string())),
@@ -124,6 +126,75 @@ async fn thread_metadata_update_patches_git_branch_and_returns_updated_thread()
Ok(())
}
#[tokio::test]
async fn thread_metadata_update_replaces_metadata_and_returns_updated_thread() -> 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 mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let initial_metadata = BTreeMap::from([
("client".to_string(), "desktop".to_string()),
("surface".to_string(), "composer".to_string()),
]);
let start_id = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
metadata: Some(initial_metadata),
..Default::default()
})
.await?;
let start_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let updated_metadata = BTreeMap::from([("client".to_string(), "vscode".to_string())]);
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread.id.clone(),
metadata: Some(updated_metadata.clone()),
git_info: None,
})
.await?;
let update_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(update_id)),
)
.await??;
let ThreadMetadataUpdateResponse { thread: updated } =
to_response::<ThreadMetadataUpdateResponse>(update_resp)?;
assert_eq!(updated.id, thread.id);
assert_eq!(updated.metadata, updated_metadata);
assert_eq!(updated.git_info, None);
let read_id = mcp
.send_thread_read_request(ThreadReadParams {
thread_id: thread.id,
include_turns: false,
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread: read } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(
read.metadata,
BTreeMap::from([("client".to_string(), "vscode".to_string())])
);
assert_eq!(read.git_info, None);
Ok(())
}
#[tokio::test]
async fn thread_metadata_update_rejects_empty_git_info_patch() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -149,6 +220,7 @@ async fn thread_metadata_update_rejects_empty_git_info_patch() -> Result<()> {
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread.id,
metadata: None,
git_info: Some(ThreadMetadataGitInfoUpdateParams {
sha: None,
branch: None,
@@ -193,6 +265,7 @@ async fn thread_metadata_update_repairs_missing_sqlite_row_for_stored_thread() -
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread_id.clone(),
metadata: None,
git_info: Some(ThreadMetadataGitInfoUpdateParams {
sha: None,
branch: Some(Some("feature/stored-thread".to_string())),
@@ -273,6 +346,7 @@ async fn thread_metadata_update_repairs_loaded_thread_without_resetting_summary(
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread_id.clone(),
metadata: None,
git_info: Some(ThreadMetadataGitInfoUpdateParams {
sha: None,
branch: Some(Some("feature/loaded-thread".to_string())),
@@ -336,6 +410,7 @@ async fn thread_metadata_update_repairs_missing_sqlite_row_for_archived_thread()
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread_id.clone(),
metadata: None,
git_info: Some(ThreadMetadataGitInfoUpdateParams {
sha: None,
branch: Some(Some("feature/archived-thread".to_string())),
@@ -392,6 +467,7 @@ async fn thread_metadata_update_can_clear_stored_git_fields() -> Result<()> {
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread_id.clone(),
metadata: None,
git_info: Some(ThreadMetadataGitInfoUpdateParams {
sha: Some(None),
branch: Some(None),
codex-rs/app-server/tests/suite/v2/thread_read.rs
+72
View File
@@ -29,6 +29,7 @@ use codex_protocol::user_input::TextElement;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::collections::BTreeMap;
use std::path::Path;
use std::path::PathBuf;
use tempfile::TempDir;
@@ -201,6 +202,50 @@ async fn thread_read_loaded_thread_returns_precomputed_path_before_materializati
Ok(())
}
#[tokio::test]
async fn thread_read_returns_persisted_metadata() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_sqlite_config_toml(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let metadata = BTreeMap::from([("client".to_string(), "desktop".to_string())]);
let start_id = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
metadata: Some(metadata.clone()),
..Default::default()
})
.await?;
let start_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let read_id = mcp
.send_thread_read_request(ThreadReadParams {
thread_id: thread.id.clone(),
include_turns: false,
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread: read } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(read.id, thread.id);
assert_eq!(read.metadata, metadata);
assert_eq!(read.status, ThreadStatus::Idle);
Ok(())
}
#[tokio::test]
async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -497,3 +542,30 @@ stream_max_retries = 0
),
)
}
fn create_sqlite_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
format!(
r#"
model = "mock-model"
approval_policy = "never"
sandbox_mode = "read-only"
suppress_unstable_features_warning = true
model_provider = "mock_provider"
[features]
sqlite = true
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
"#
),
)
}
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+1
View File
@@ -376,6 +376,7 @@ stream_max_retries = 0
let update_id = mcp
.send_thread_metadata_update_request(ThreadMetadataUpdateParams {
thread_id: thread_id.clone(),
metadata: None,
git_info: Some(ThreadMetadataGitInfoUpdateParams {
sha: None,
branch: Some(Some("feature/pr-branch".to_string())),
codex-rs/app-server/tests/suite/v2/thread_start.rs
+48
View File
@@ -25,6 +25,7 @@ use codex_protocol::openai_models::ReasoningEffort;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::collections::BTreeMap;
use std::path::Path;
use tempfile::TempDir;
use tokio::time::timeout;
@@ -48,10 +49,13 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let metadata = BTreeMap::from([("client".to_string(), "vscode".to_string())]);
// Start a v2 thread with an explicit model override.
let req_id = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("gpt-5.1".to_string()),
metadata: Some(metadata.clone()),
..Default::default()
})
.await?;
@@ -83,6 +87,7 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
"new persistent threads should not be ephemeral"
);
assert_eq!(thread.status, ThreadStatus::Idle);
assert_eq!(thread.metadata, metadata);
let thread_path = thread.path.clone().expect("thread path should be present");
assert!(thread_path.is_absolute(), "thread path should be absolute");
assert!(
@@ -105,6 +110,11 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
Some(false),
"new persistent threads should serialize `ephemeral: false`"
);
assert_eq!(
thread_json.get("metadata"),
Some(&json!({ "client": "vscode" })),
"new threads should serialize client metadata"
);
assert_eq!(thread.name, None);
// A corresponding thread/started notification should arrive.
@@ -146,6 +156,11 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
Some(false),
"thread/started should serialize `ephemeral: false` for new persistent threads"
);
assert_eq!(
started_thread_json.get("metadata"),
Some(&json!({ "client": "vscode" })),
"thread/started should serialize client metadata"
);
let started: ThreadStartedNotification =
serde_json::from_value(notif.params.expect("params must be present"))?;
assert_eq!(started.thread, thread);
@@ -295,6 +310,39 @@ async fn thread_start_ephemeral_remains_pathless() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_start_rejects_metadata_over_limit() -> 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 mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let metadata = (0..17)
.map(|idx| (format!("key-{idx}"), format!("value-{idx}")))
.collect();
let req_id = mcp
.send_thread_start_request(ThreadStartParams {
metadata: Some(metadata),
..Default::default()
})
.await?;
let err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(req_id)),
)
.await??;
assert_eq!(
err.error.message,
"metadata must include at most 16 entries"
);
Ok(())
}
#[tokio::test]
async fn thread_start_fails_when_required_mcp_server_fails_to_initialize() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
codex-rs/core/src/realtime_context_tests.rs
+1
View File
@@ -40,6 +40,7 @@ fn thread_metadata(cwd: &str, title: &str, first_user_message: &str) -> ThreadMe
git_sha: None,
git_branch: Some("main".to_string()),
git_origin_url: None,
metadata_json: "{}".to_string(),
}
}
codex-rs/exec/src/lib.rs
+2
View File
@@ -1837,6 +1837,7 @@ mod tests {
agent_role: None,
git_info: None,
name: None,
metadata: std::collections::BTreeMap::new(),
turns: vec![
codex_app_server_protocol::Turn {
id: "turn-1".to_string(),
@@ -1952,6 +1953,7 @@ mod tests {
agent_role: None,
git_info: None,
name: Some("thread".to_string()),
metadata: std::collections::BTreeMap::new(),
turns: vec![],
},
model: "gpt-5.4".to_string(),
codex-rs/state/migrations/0023_threads_metadata_json.sql
+1
View File
@@ -0,0 +1 @@
ALTER TABLE threads ADD COLUMN metadata_json TEXT NOT NULL DEFAULT '{}';
codex-rs/state/src/extract.rs
+1
View File
@@ -422,6 +422,7 @@ mod tests {
git_sha: None,
git_branch: None,
git_origin_url: None,
metadata_json: "{}".to_string(),
}
}
codex-rs/state/src/model/thread_metadata.rs
+15
View File
@@ -99,6 +99,8 @@ pub struct ThreadMetadata {
pub git_branch: Option<String>,
/// The git origin URL, if known.
pub git_origin_url: Option<String>,
/// Arbitrary client-defined thread metadata stored as canonical JSON.
pub metadata_json: String,
}
/// Builder data required to construct [`ThreadMetadata`] without parsing filenames.
@@ -138,6 +140,8 @@ pub struct ThreadMetadataBuilder {
pub git_branch: Option<String>,
/// The git origin URL, if known.
pub git_origin_url: Option<String>,
/// Arbitrary client-defined thread metadata stored as canonical JSON.
pub metadata_json: String,
}
impl ThreadMetadataBuilder {
@@ -166,6 +170,7 @@ impl ThreadMetadataBuilder {
git_sha: None,
git_branch: None,
git_origin_url: None,
metadata_json: "{}".to_string(),
}
}
@@ -208,6 +213,7 @@ impl ThreadMetadataBuilder {
git_sha: self.git_sha.clone(),
git_branch: self.git_branch.clone(),
git_origin_url: self.git_origin_url.clone(),
metadata_json: self.metadata_json.clone(),
}
}
}
@@ -295,6 +301,9 @@ impl ThreadMetadata {
if self.git_origin_url != other.git_origin_url {
diffs.push("git_origin_url");
}
if self.metadata_json != other.metadata_json {
diffs.push("metadata_json");
}
diffs
}
}
@@ -327,6 +336,7 @@ pub(crate) struct ThreadRow {
git_sha: Option<String>,
git_branch: Option<String>,
git_origin_url: Option<String>,
metadata_json: String,
}
impl ThreadRow {
@@ -354,6 +364,7 @@ impl ThreadRow {
git_sha: row.try_get("git_sha")?,
git_branch: row.try_get("git_branch")?,
git_origin_url: row.try_get("git_origin_url")?,
metadata_json: row.try_get("metadata_json")?,
})
}
}
@@ -385,6 +396,7 @@ impl TryFrom<ThreadRow> for ThreadMetadata {
git_sha,
git_branch,
git_origin_url,
metadata_json,
} = row;
Ok(Self {
id: ThreadId::try_from(id)?,
@@ -410,6 +422,7 @@ impl TryFrom<ThreadRow> for ThreadMetadata {
git_sha,
git_branch,
git_origin_url,
metadata_json,
})
}
}
@@ -478,6 +491,7 @@ mod tests {
git_sha: None,
git_branch: None,
git_origin_url: None,
metadata_json: "{}".to_string(),
}
}
@@ -506,6 +520,7 @@ mod tests {
git_sha: None,
git_branch: None,
git_origin_url: None,
metadata_json: "{}".to_string(),
}
}
codex-rs/state/src/runtime/memories.rs
+2 -1
View File
@@ -182,7 +182,8 @@ SELECT
archived_at,
git_sha,
git_branch,
git_origin_url
git_origin_url,
metadata_json
FROM threads
LEFT JOIN stage1_outputs
ON stage1_outputs.thread_id = threads.id
codex-rs/state/src/runtime/test_support.rs
+1
View File
@@ -65,5 +65,6 @@ pub(super) fn test_thread_metadata(
git_sha: None,
git_branch: None,
git_origin_url: None,
metadata_json: "{}".to_string(),
}
}
codex-rs/state/src/runtime/threads.rs
+25 -5
View File
@@ -27,7 +27,8 @@ SELECT
archived_at,
git_sha,
git_branch,
git_origin_url
git_origin_url,
metadata_json
FROM threads
WHERE id = ?
"#,
@@ -364,7 +365,8 @@ SELECT
archived_at,
git_sha,
git_branch,
git_origin_url
git_origin_url,
metadata_json
FROM threads
"#,
);
@@ -467,8 +469,9 @@ INSERT INTO threads (
git_sha,
git_branch,
git_origin_url,
metadata_json,
memory_mode
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO NOTHING
"#,
)
@@ -500,6 +503,7 @@ ON CONFLICT(id) DO NOTHING
.bind(metadata.git_sha.as_deref())
.bind(metadata.git_branch.as_deref())
.bind(metadata.git_origin_url.as_deref())
.bind(metadata.metadata_json.as_str())
.bind("enabled")
.execute(self.pool.as_ref())
.await?;
@@ -563,6 +567,19 @@ WHERE id = ?
Ok(result.rows_affected() > 0)
}
pub async fn update_thread_metadata_json(
&self,
thread_id: ThreadId,
metadata_json: &str,
) -> anyhow::Result<bool> {
let result = sqlx::query("UPDATE threads SET metadata_json = ? WHERE id = ?")
.bind(metadata_json)
.bind(thread_id.to_string())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
async fn upsert_thread_with_creation_memory_mode(
&self,
metadata: &crate::ThreadMetadata,
@@ -594,8 +611,9 @@ INSERT INTO threads (
git_sha,
git_branch,
git_origin_url,
metadata_json,
memory_mode
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
rollout_path = excluded.rollout_path,
created_at = excluded.created_at,
@@ -618,7 +636,8 @@ ON CONFLICT(id) DO UPDATE SET
archived_at = excluded.archived_at,
git_sha = excluded.git_sha,
git_branch = excluded.git_branch,
git_origin_url = excluded.git_origin_url
git_origin_url = excluded.git_origin_url,
metadata_json = excluded.metadata_json
"#,
)
.bind(metadata.id.to_string())
@@ -649,6 +668,7 @@ ON CONFLICT(id) DO UPDATE SET
.bind(metadata.git_sha.as_deref())
.bind(metadata.git_branch.as_deref())
.bind(metadata.git_origin_url.as_deref())
.bind(metadata.metadata_json.as_str())
.bind(creation_memory_mode.unwrap_or("enabled"))
.execute(self.pool.as_ref())
.await?;
codex-rs/tui_app_server/src/app.rs
+3
View File
@@ -7418,6 +7418,7 @@ guardian_approval = true
agent_role: Some("explorer".to_string()),
git_info: None,
name: Some("agent thread".to_string()),
metadata: std::collections::BTreeMap::new(),
turns: Vec::new(),
},
}),
@@ -7494,6 +7495,7 @@ guardian_approval = true
agent_role: Some("explorer".to_string()),
git_info: None,
name: Some("agent thread".to_string()),
metadata: std::collections::BTreeMap::new(),
turns: Vec::new(),
},
}),
@@ -9080,6 +9082,7 @@ guardian_approval = true
agent_role: None,
git_info: None,
name: None,
metadata: std::collections::BTreeMap::new(),
turns: Vec::new(),
},
},
codex-rs/tui_app_server/src/app/app_server_adapter.rs
+2
View File
@@ -1262,6 +1262,7 @@ mod tests {
agent_role: None,
git_info: None,
name: None,
metadata: std::collections::BTreeMap::new(),
turns: vec![Turn {
id: "turn-1".to_string(),
items: vec![ThreadItem::CommandExecution {
@@ -1428,6 +1429,7 @@ mod tests {
agent_role: None,
git_info: None,
name: Some("restore".to_string()),
metadata: std::collections::BTreeMap::new(),
turns: vec![
Turn {
id: "turn-complete".to_string(),
codex-rs/tui_app_server/src/app_server_session.rs
+1
View File
@@ -1157,6 +1157,7 @@ mod tests {
agent_role: None,
git_info: None,
name: None,
metadata: std::collections::BTreeMap::new(),
turns: vec![Turn {
id: "turn-1".to_string(),
items: vec![
codex-rs/tui_app_server/src/resume_picker.rs
+1
View File
@@ -2574,6 +2574,7 @@ mod tests {
agent_role: None,
git_info: None,
name: Some(String::from("Named thread")),
metadata: std::collections::BTreeMap::new(),
turns: Vec::new(),
};