openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into etraut/loop

Browse Source
This commit is contained in:
Eric Traut
2026-03-31 16:32:35 -06:00
committed by GitHub
parent 7bca5b7a06 9a8730f31e
commit 24b097484d
59 changed files with 2169 additions and 907 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.codex/skills/babysit-pr/SKILL.md
+13 -14
View File
@@ -1,6 +1,6 @@
---
name: babysit-pr
description: Babysit a GitHub pull request after creation by continuously polling CI checks/workflow runs, new review comments, and mergeability state until the PR is ready to merge (or merged/closed). Diagnose failures, retry likely flaky failures up to 3 times, auto-fix/push branch-related issues when appropriate, and stop only when user help is required (for example CI infrastructure issues, exhausted flaky retries, or ambiguous/blocking situations). Use when the user asks Codex to monitor a PR, watch CI, handle review comments, or keep an eye on failures and feedback on an open PR.
description: Babysit a GitHub pull request after creation by continuously polling review comments, CI checks/workflow runs, and mergeability state until the PR is merged/closed or user help is required. Diagnose failures, retry likely flaky failures up to 3 times, auto-fix/push branch-related issues when appropriate, and keep watching open PRs so fresh review feedback is surfaced promptly. Use when the user asks Codex to monitor a PR, watch CI, handle review comments, or keep an eye on failures and feedback on an open PR.
---
# PR Babysitter
@@ -9,8 +9,8 @@ description: Babysit a GitHub pull request after creation by continuously pollin
Babysit a PR persistently until one of these terminal outcomes occurs:
- The PR is merged or closed.
- CI is successful, there are no unaddressed review comments surfaced by the watcher, required review approval is not blocking merge, and there are no potential merge conflicts (PR is mergeable / not reporting conflict risk).
- A situation requires user help (for example CI infrastructure issues, repeated flaky failures after retry budget is exhausted, permission problems, or ambiguity that cannot be resolved safely).
- Optional handoff milestone: the PR is currently green + mergeable + review-clean. Treat this as a progress state, not a watcher stop, so late-arriving review comments are still surfaced promptly while the PR remains open.
Do not stop merely because a single snapshot returns `idle` while checks are still pending.
@@ -24,7 +24,7 @@ Accept any of the following:
## Core Workflow
1. When the user asks to "monitor"/"watch"/"babysit" a PR, start with the watcher's continuous mode (`--watch`) unless you are intentionally doing a one-shot diagnostic snapshot.
2. Run the watcher script to snapshot PR/CI/review state (or consume each streamed snapshot from `--watch`).
2. Run the watcher script to snapshot PR/review/CI state (or consume each streamed snapshot from `--watch`).
3. Inspect the `actions` list in the JSON response.
4. If `diagnose_ci_failure` is present, inspect failed run logs and classify the failure.
5. If the failure is likely caused by the current branch, patch code locally, commit, and push.
@@ -33,10 +33,10 @@ Accept any of the following:
8. If a review item from another author is non-actionable, already addressed, or not valid, post one reply on the comment/thread explaining that decision (for example answering the question or explaining why no change is needed). If the watcher later surfaces your own reply, treat that self-authored item as already handled and do not reply again.
9. If the failure is likely flaky/unrelated and `retry_failed_checks` is present, rerun failed jobs with `--retry-failed-now`.
10. If both actionable review feedback and `retry_failed_checks` are present, prioritize review feedback first; a new commit will retrigger CI, so avoid rerunning flaky checks on the old SHA unless you intentionally defer the review change.
11. On every loop, verify mergeability / merge-conflict status (for example via `gh pr view`) in addition to CI and review state.
11. On every loop, look for newly surfaced review feedback before acting on CI failures or mergeability state, then verify mergeability / merge-conflict status (for example via `gh pr view`) alongside CI.
12. After any push or rerun action, immediately return to step 1 and continue polling on the updated SHA/state.
13. If you had been using `--watch` before pausing to patch/commit/push, relaunch `--watch` yourself in the same turn immediately after the push (do not wait for the user to re-invoke the skill).
14. Repeat polling until the PR is green + review-clean + mergeable, `stop_pr_closed` appears, or a user-help-required blocker is reached.
14. Repeat polling until `stop_pr_closed` appears or a user-help-required blocker is reached. A green + review-clean + mergeable PR is a progress milestone, not a reason to stop the watcher while the PR is still open.
15. Maintain terminal/session ownership: while babysitting is active, keep consuming watcher output in the same turn; do not leave a detached `--watch` process running and then end the turn as if monitoring were complete.
## Commands
@@ -131,7 +131,7 @@ Use this loop in a live Codex session:
8. Retry failed checks only when `retry_failed_checks` is present and you are not about to replace the current SHA with a review/CI fix commit.
9. If you pushed a commit, resolved a review thread, replied to a review comment, or triggered a rerun, report the action briefly and continue polling (do not stop).
10. After a review-fix push, proactively restart continuous monitoring (`--watch`) in the same turn unless a strict stop condition has already been reached.
11. If everything is passing, mergeable, not blocked on required review approval, and there are no unaddressed review items, report success and stop.
11. If everything is passing, mergeable, not blocked on required review approval, and there are no unaddressed review items, report that the PR is currently ready to merge but keep the watcher running so new review comments are surfaced quickly while the PR remains open.
12. If blocked on a user-help-required issue (infra outage, exhausted flaky retries, unclear reviewer request, permissions), report the blocker and stop.
13. Otherwise sleep according to the polling cadence below and repeat.
@@ -141,19 +141,18 @@ Do not hand control back to the user after a review-fix push just because a new
If a `--watch` process is still running and no strict stop condition has been reached, the babysitting task is still in progress; keep streaming/consuming watcher output instead of ending the turn.
## Polling Cadence
Use adaptive polling and continue monitoring even after CI turns green:
Keep review polling aggressive and continue monitoring even after CI turns green:
- While CI is not green (pending/running/queued or failing): poll every 1 minute.
- After CI turns green: start at every 1 minute, then back off exponentially when there is no change (for example 1m, 2m, 4m, 8m, 16m, 32m), capping at every 1 hour.
- Reset the green-state polling interval back to 1 minute whenever anything changes (new commit/SHA, check status changes, new review comments, mergeability changes, review decision changes).
- If CI stops being green again (new commit, rerun, or regression): return to 1-minute polling.
- After CI turns green: keep polling at the base cadence while the PR remains open so newly posted review comments are surfaced promptly instead of waiting on a long green-state backoff.
- Reset the cadence immediately whenever anything changes (new commit/SHA, check status changes, new review comments, mergeability changes, review decision changes).
- If CI stops being green again (new commit, rerun, or regression): stay on the base polling cadence.
- If any poll shows the PR is merged or otherwise closed: stop polling immediately and report the terminal state.
## Stop Conditions (Strict)
Stop only when one of the following is true:
- PR merged or closed (stop as soon as a poll/snapshot confirms this).
- PR is ready to merge: CI succeeded, no surfaced unaddressed review comments, not blocked on required review approval, and no merge conflict risk.
- User intervention is required and Codex cannot safely proceed alone.
Keep polling when:
@@ -162,14 +161,14 @@ Keep polling when:
- CI is still running/queued.
- Review state is quiet but CI is not terminal.
- CI is green but mergeability is unknown/pending.
- CI is green and mergeable, but the PR is still open and you are waiting for possible new review comments or merge-conflict changes per the green-state cadence.
- The PR is green but blocked on review approval (`REVIEW_REQUIRED` / similar); continue polling on the green-state cadence and surface any new review comments without asking for confirmation to keep watching.
- CI is green and mergeable, but the PR is still open and you are waiting for possible new review comments or merge-conflict changes.
- The PR is green but blocked on review approval (`REVIEW_REQUIRED` / similar); continue polling at the base cadence and surface any new review comments without asking for confirmation to keep watching.
## Output Expectations
Provide concise progress updates while monitoring and a final summary that includes:
- During long unchanged monitoring periods, avoid emitting a full update on every poll; summarize only status changes plus occasional heartbeat updates.
- Treat push confirmations, intermediate CI snapshots, and review-action updates as progress updates only; do not emit the final summary or end the babysitting session unless a strict stop condition is met.
- Treat push confirmations, intermediate CI snapshots, ready-to-merge snapshots, and review-action updates as progress updates only; do not emit the final summary or end the babysitting session unless a strict stop condition is met.
- A user request to "monitor" is not satisfied by a couple of sample polls; remain in the loop until a strict stop condition or an explicit user interruption.
- A review-fix commit + push is not a completion event; immediately resume live monitoring (`--watch`) in the same turn and continue reporting progress updates.
- When CI first transitions to all green for the current SHA, emit a one-time celebratory progress update (do not repeat it on every green poll). Preferred style: `🚀 CI is all green! 33/33 passed. Still on watch for review approval.`
.codex/skills/babysit-pr/agents/openai.yaml
+2 -2
View File
@@ -1,4 +1,4 @@
interface:
display_name: "PR Babysitter"
short_description: "Watch PR CI, reviews, and merge conflicts"
default_prompt: "Babysit the current PR: monitor CI, reviewer comments, and merge-conflict status (prefer the watcher’s --watch mode for live monitoring); fix valid issues, push updates, and rerun flaky failures up to 3 times. Keep exactly one watcher session active for the PR (do not leave duplicate --watch terminals running). If you pause monitoring to patch review/CI feedback, restart --watch yourself immediately after the push in the same turn. If a watcher is still running and no strict stop condition has been reached, the task is still in progress: keep consuming watcher output and sending progress updates instead of ending the turn. Continue polling autonomously after any push/rerun until a strict terminal stop condition is reached or the user interrupts."
short_description: "Watch PR review comments, CI, and merge conflicts"
default_prompt: "Babysit the current PR: monitor reviewer comments, CI, and merge-conflict status (prefer the watcher’s --watch mode for live monitoring); surface new review feedback before acting on CI or mergeability work, fix valid issues, push updates, and rerun flaky failures up to 3 times. Keep exactly one watcher session active for the PR (do not leave duplicate --watch terminals running). If you pause monitoring to patch review/CI feedback, restart --watch yourself immediately after the push in the same turn. If a watcher is still running and no strict stop condition has been reached, the task is still in progress: keep consuming watcher output and sending progress updates instead of ending the turn. Do not treat a green + mergeable PR as a terminal stop while it is still open; continue polling autonomously after any push/rerun so newly posted review comments are surfaced until a strict terminal stop condition is reached or the user interrupts."
.codex/skills/babysit-pr/scripts/gh_pr_watch.py
+13 -12
View File
@@ -45,7 +45,6 @@ MERGE_CONFLICT_OR_BLOCKING_STATES = {
"DRAFT",
"UNKNOWN",
}
GREEN_STATE_MAX_POLL_SECONDS = 60 * 60
class GhCommandError(RuntimeError):
@@ -578,7 +577,7 @@ def recommend_actions(pr, checks_summary, failed_runs, new_review_items, retries
return unique_actions(actions)
if is_pr_ready_to_merge(pr, checks_summary, new_review_items):
actions.append("stop_ready_to_merge")
actions.append("ready_to_merge")
return unique_actions(actions)
if new_review_items:
@@ -606,12 +605,6 @@ def collect_snapshot(args):
if not state.get("started_at"):
state["started_at"] = int(time.time())
# `gh pr checks -R <repo>` requires an explicit PR/branch/url argument.
# After resolving `--pr auto`, reuse the concrete PR number.
checks = get_pr_checks(str(pr["number"]), repo=pr["repo"])
checks_summary = summarize_checks(checks)
workflow_runs = get_workflow_runs_for_sha(pr["repo"], pr["head_sha"])
failed_runs = failed_runs_from_workflow_runs(workflow_runs, pr["head_sha"])
authenticated_login = get_authenticated_login()
new_review_items = fetch_new_review_items(
pr,
@@ -619,6 +612,15 @@ def collect_snapshot(args):
fresh_state=fresh_state,
authenticated_login=authenticated_login,
)
# Surface review feedback before drilling into CI and mergeability details.
# That keeps the babysitter responsive to new comments even when other
# actions are also available.
# `gh pr checks -R <repo>` requires an explicit PR/branch/url argument.
# After resolving `--pr auto`, reuse the concrete PR number.
checks = get_pr_checks(str(pr["number"]), repo=pr["repo"])
checks_summary = summarize_checks(checks)
workflow_runs = get_workflow_runs_for_sha(pr["repo"], pr["head_sha"])
failed_runs = failed_runs_from_workflow_runs(workflow_runs, pr["head_sha"])
retries_used = current_retry_count(state, pr["head_sha"])
actions = recommend_actions(
@@ -761,7 +763,6 @@ def run_watch(args):
if (
"stop_pr_closed" in actions
or "stop_exhausted_retries" in actions
or "stop_ready_to_merge" in actions
):
print_event("stop", {"actions": snapshot.get("actions"), "pr": snapshot.get("pr")})
return 0
@@ -769,13 +770,13 @@ def run_watch(args):
current_change_key = snapshot_change_key(snapshot)
changed = current_change_key != last_change_key
green = is_ci_green(snapshot)
pr = snapshot.get("pr") or {}
pr_open = not bool(pr.get("closed")) and not bool(pr.get("merged"))
if not green:
if not green or pr_open:
poll_seconds = args.poll_seconds
elif changed or last_change_key is None:
poll_seconds = args.poll_seconds
else:
poll_seconds = min(poll_seconds * 2, GREEN_STATE_MAX_POLL_SECONDS)
last_change_key = current_change_key
time.sleep(poll_seconds)
.codex/skills/babysit-pr/scripts/test_gh_pr_watch.py
+155
View File
@@ -0,0 +1,155 @@
import argparse
import importlib.util
from pathlib import Path
import pytest
MODULE_PATH = Path(__file__).with_name("gh_pr_watch.py")
MODULE_SPEC = importlib.util.spec_from_file_location("gh_pr_watch", MODULE_PATH)
gh_pr_watch = importlib.util.module_from_spec(MODULE_SPEC)
assert MODULE_SPEC.loader is not None
MODULE_SPEC.loader.exec_module(gh_pr_watch)
def sample_pr():
return {
"number": 123,
"url": "https://github.com/openai/codex/pull/123",
"repo": "openai/codex",
"head_sha": "abc123",
"head_branch": "feature",
"state": "OPEN",
"merged": False,
"closed": False,
"mergeable": "MERGEABLE",
"merge_state_status": "CLEAN",
"review_decision": "",
}
def sample_checks(**overrides):
checks = {
"pending_count": 0,
"failed_count": 0,
"passed_count": 12,
"all_terminal": True,
}
checks.update(overrides)
return checks
def test_collect_snapshot_fetches_review_items_before_ci(monkeypatch, tmp_path):
call_order = []
pr = sample_pr()
monkeypatch.setattr(gh_pr_watch, "resolve_pr", lambda *args, **kwargs: pr)
monkeypatch.setattr(gh_pr_watch, "load_state", lambda path: ({}, True))
monkeypatch.setattr(
gh_pr_watch,
"get_authenticated_login",
lambda: call_order.append("auth") or "octocat",
)
monkeypatch.setattr(
gh_pr_watch,
"fetch_new_review_items",
lambda *args, **kwargs: call_order.append("review") or [],
)
monkeypatch.setattr(
gh_pr_watch,
"get_pr_checks",
lambda *args, **kwargs: call_order.append("checks") or [],
)
monkeypatch.setattr(
gh_pr_watch,
"summarize_checks",
lambda checks: call_order.append("summarize") or sample_checks(),
)
monkeypatch.setattr(
gh_pr_watch,
"get_workflow_runs_for_sha",
lambda *args, **kwargs: call_order.append("workflow") or [],
)
monkeypatch.setattr(
gh_pr_watch,
"failed_runs_from_workflow_runs",
lambda *args, **kwargs: call_order.append("failed_runs") or [],
)
monkeypatch.setattr(
gh_pr_watch,
"recommend_actions",
lambda *args, **kwargs: call_order.append("recommend") or ["idle"],
)
monkeypatch.setattr(gh_pr_watch, "save_state", lambda *args, **kwargs: None)
args = argparse.Namespace(
pr="123",
repo=None,
state_file=str(tmp_path / "watcher-state.json"),
max_flaky_retries=3,
)
gh_pr_watch.collect_snapshot(args)
assert call_order.index("review") < call_order.index("checks")
assert call_order.index("review") < call_order.index("workflow")
def test_recommend_actions_prioritizes_review_comments():
actions = gh_pr_watch.recommend_actions(
sample_pr(),
sample_checks(failed_count=1),
[{"run_id": 99}],
[{"kind": "review_comment", "id": "1"}],
0,
3,
)
assert actions == [
"process_review_comment",
"diagnose_ci_failure",
"retry_failed_checks",
]
def test_run_watch_keeps_polling_open_ready_to_merge_pr(monkeypatch):
sleeps = []
events = []
snapshot = {
"pr": sample_pr(),
"checks": sample_checks(),
"failed_runs": [],
"new_review_items": [],
"actions": ["ready_to_merge"],
"retry_state": {
"current_sha_retries_used": 0,
"max_flaky_retries": 3,
},
}
monkeypatch.setattr(
gh_pr_watch,
"collect_snapshot",
lambda args: (snapshot, Path("/tmp/codex-babysit-pr-state.json")),
)
monkeypatch.setattr(
gh_pr_watch,
"print_event",
lambda event, payload: events.append((event, payload)),
)
class StopWatch(Exception):
pass
def fake_sleep(seconds):
sleeps.append(seconds)
if len(sleeps) >= 2:
raise StopWatch
monkeypatch.setattr(gh_pr_watch.time, "sleep", fake_sleep)
with pytest.raises(StopWatch):
gh_pr_watch.run_watch(argparse.Namespace(poll_seconds=30))
assert sleeps == [30, 30]
assert [event for event, _ in events] == ["snapshot", "snapshot"]
.github/scripts/verify_cargo_workspace_manifests.py
+125
View File
@@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""Verify that codex-rs crates inherit workspace metadata, lints, and names.
This keeps `cargo clippy` aligned with the workspace lint policy by ensuring
each crate opts into `[lints] workspace = true`, and it also checks the crate
name conventions for top-level `codex-rs/*` crates and `codex-rs/utils/*`
crates.
"""
from __future__ import annotations
import sys
import tomllib
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
CARGO_RS_ROOT = ROOT / "codex-rs"
WORKSPACE_PACKAGE_FIELDS = ("version", "edition", "license")
TOP_LEVEL_NAME_EXCEPTIONS = {
"windows-sandbox-rs": "codex-windows-sandbox",
}
UTILITY_NAME_EXCEPTIONS = {
"path-utils": "codex-utils-path",
}
def main() -> int:
failures = [
(path.relative_to(ROOT), errors)
for path in cargo_manifests()
if (errors := manifest_errors(path))
]
if not failures:
return 0
print(
"Cargo manifests under codex-rs must inherit workspace package metadata and "
"opt into workspace lints."
)
print(
"Cargo only applies `codex-rs/Cargo.toml` `[workspace.lints.clippy]` "
"entries to a crate when that crate declares:"
)
print()
print("[lints]")
print("workspace = true")
print()
print(
"Without that opt-in, `cargo clippy` can miss violations that Bazel clippy "
"catches."
)
print()
print(
"Package-name checks apply to `codex-rs/<crate>/Cargo.toml` and "
"`codex-rs/utils/<crate>/Cargo.toml`."
)
print()
for path, errors in failures:
print(f"{path}:")
for error in errors:
print(f" - {error}")
return 1
def manifest_errors(path: Path) -> list[str]:
manifest = load_manifest(path)
package = manifest.get("package")
if not isinstance(package, dict):
return []
errors = []
for field in WORKSPACE_PACKAGE_FIELDS:
if not is_workspace_reference(package.get(field)):
errors.append(f"set `{field}.workspace = true` in `[package]`")
lints = manifest.get("lints")
if not (isinstance(lints, dict) and lints.get("workspace") is True):
errors.append("add `[lints]` with `workspace = true`")
expected_name = expected_package_name(path)
if expected_name is not None:
actual_name = package.get("name")
if actual_name != expected_name:
errors.append(
f"set `[package].name` to `{expected_name}` (found `{actual_name}`)"
)
return errors
def expected_package_name(path: Path) -> str | None:
parts = path.relative_to(CARGO_RS_ROOT).parts
if len(parts) == 2 and parts[1] == "Cargo.toml":
directory = parts[0]
return TOP_LEVEL_NAME_EXCEPTIONS.get(
directory,
directory if directory.startswith("codex-") else f"codex-{directory}",
)
if len(parts) == 3 and parts[0] == "utils" and parts[2] == "Cargo.toml":
directory = parts[1]
return UTILITY_NAME_EXCEPTIONS.get(directory, f"codex-utils-{directory}")
return None
def is_workspace_reference(value: object) -> bool:
return isinstance(value, dict) and value.get("workspace") is True
def load_manifest(path: Path) -> dict:
return tomllib.loads(path.read_text())
def cargo_manifests() -> list[Path]:
return sorted(
path
for path in CARGO_RS_ROOT.rglob("Cargo.toml")
if path != CARGO_RS_ROOT / "Cargo.toml"
)
if __name__ == "__main__":
sys.exit(main())
.github/workflows/ci.yml
+3
View File
@@ -14,6 +14,9 @@ jobs:
- name: Checkout repository
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
- name: Verify codex-rs Cargo manifests inherit workspace settings
run: python3 .github/scripts/verify_cargo_workspace_manifests.py
- name: Setup pnpm
uses: pnpm/action-setup@a8198c4bff370c8506180b035930dea56dbd5288 # v5
with:
AGENTS.md
+1
View File
@@ -17,6 +17,7 @@ In the codex-rs folder where the rust code lives:
- Do not add these comments for string or char literals unless the comment adds real clarity; those literals are intentionally exempt from the lint.
- If you add one of these comments, the parameter name must exactly match the callee signature.
- When possible, make `match` statements exhaustive and avoid wildcard arms.
- Newly added traits should include doc comments that explain their role and how implementations are expected to use them.
- When writing tests, prefer comparing the equality of entire objects over fields one by one.
- When making a change that adds or changes an API, ensure that the documentation in the `docs/` folder is up to date if applicable.
- If you change `ConfigToml` or nested config types, run `just write-config-schema` to update `codex-rs/core/config.schema.json`.
codex-rs/Cargo.lock
Generated
+1
View File
@@ -1342,6 +1342,7 @@ dependencies = [
"codex-login",
"codex-plugin",
"codex-protocol",
"os_info",
"pretty_assertions",
"serde",
"serde_json",
codex-rs/analytics/Cargo.toml
+1
View File
@@ -18,6 +18,7 @@ codex-git-utils = { workspace = true }
codex-login = { workspace = true }
codex-plugin = { workspace = true }
codex-protocol = { workspace = true }
os_info = { workspace = true }
serde = { workspace = true, features = ["derive"] }
sha1 = { workspace = true }
tokio = { workspace = true, features = [
codex-rs/analytics/src/analytics_client.rs
-671
View File
@@ -1,671 +0,0 @@
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_git_utils::collect_git_info;
use codex_git_utils::get_git_repo_root;
use codex_login::AuthManager;
use codex_login::default_client::create_client;
use codex_login::default_client::originator;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::protocol::SkillScope;
use serde::Serialize;
use sha1::Digest;
use sha1::Sha1;
use std::collections::HashSet;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use tokio::sync::mpsc;
#[derive(Clone)]
pub struct TrackEventsContext {
pub model_slug: String,
pub thread_id: String,
pub turn_id: String,
}
pub fn build_track_events_context(
model_slug: String,
thread_id: String,
turn_id: String,
) -> TrackEventsContext {
TrackEventsContext {
model_slug,
thread_id,
turn_id,
}
}
#[derive(Clone, Debug)]
pub struct SkillInvocation {
pub skill_name: String,
pub skill_scope: SkillScope,
pub skill_path: PathBuf,
pub invocation_type: InvocationType,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum InvocationType {
Explicit,
Implicit,
}
pub struct AppInvocation {
pub connector_id: Option<String>,
pub app_name: Option<String>,
pub invocation_type: Option<InvocationType>,
}
pub enum AnalyticsFact {
Initialize {
connection_id: u64,
params: InitializeParams,
},
Request {
connection_id: u64,
request_id: RequestId,
request: Box<ClientRequest>,
},
Response {
connection_id: u64,
response: Box<ClientResponse>,
},
Notification(Box<ServerNotification>),
// Facts that do not naturally exist on the app-server protocol surface, or
// would require non-trivial protocol reshaping on this branch.
Custom(CustomAnalyticsFact),
}
pub enum CustomAnalyticsFact {
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
AppUsed(AppUsedInput),
PluginUsed(PluginUsedInput),
PluginStateChanged(PluginStateChangedInput),
}
pub struct SkillInvokedInput {
pub tracking: TrackEventsContext,
pub invocations: Vec<SkillInvocation>,
}
pub struct AppMentionedInput {
pub tracking: TrackEventsContext,
pub mentions: Vec<AppInvocation>,
}
pub struct AppUsedInput {
pub tracking: TrackEventsContext,
pub app: AppInvocation,
}
pub struct PluginUsedInput {
pub tracking: TrackEventsContext,
pub plugin: PluginTelemetryMetadata,
}
pub struct PluginStateChangedInput {
pub plugin: PluginTelemetryMetadata,
pub state: PluginState,
}
#[derive(Clone, Copy)]
pub enum PluginState {
Installed,
Uninstalled,
Enabled,
Disabled,
}
#[derive(Default)]
pub struct AnalyticsReducer;
#[derive(Clone)]
pub(crate) struct AnalyticsEventsQueue {
sender: mpsc::Sender<AnalyticsFact>,
app_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
plugin_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
}
#[derive(Clone)]
pub struct AnalyticsEventsClient {
queue: AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
}
impl AnalyticsEventsQueue {
pub(crate) fn new(auth_manager: Arc<AuthManager>, base_url: String) -> Self {
let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE);
tokio::spawn(async move {
let mut reducer = AnalyticsReducer;
while let Some(input) = receiver.recv().await {
let mut events = Vec::new();
reducer.ingest(input, &mut events).await;
send_track_events(&auth_manager, &base_url, events).await;
}
});
Self {
sender,
app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
}
}
fn try_send(&self, input: AnalyticsFact) {
if self.sender.try_send(input).is_err() {
//TODO: add a metric for this
tracing::warn!("dropping analytics events: queue is full");
}
}
fn should_enqueue_app_used(&self, tracking: &TrackEventsContext, app: &AppInvocation) -> bool {
let Some(connector_id) = app.connector_id.as_ref() else {
return true;
};
let mut emitted = self
.app_used_emitted_keys
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if emitted.len() >= ANALYTICS_EVENT_DEDUPE_MAX_KEYS {
emitted.clear();
}
emitted.insert((tracking.turn_id.clone(), connector_id.clone()))
}
fn should_enqueue_plugin_used(
&self,
tracking: &TrackEventsContext,
plugin: &PluginTelemetryMetadata,
) -> bool {
let mut emitted = self
.plugin_used_emitted_keys
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if emitted.len() >= ANALYTICS_EVENT_DEDUPE_MAX_KEYS {
emitted.clear();
}
emitted.insert((tracking.turn_id.clone(), plugin.plugin_id.as_key()))
}
}
impl AnalyticsEventsClient {
pub fn new(
auth_manager: Arc<AuthManager>,
base_url: String,
analytics_enabled: Option<bool>,
) -> Self {
Self {
queue: AnalyticsEventsQueue::new(Arc::clone(&auth_manager), base_url),
analytics_enabled,
}
}
pub fn track_skill_invocations(
&self,
tracking: TrackEventsContext,
invocations: Vec<SkillInvocation>,
) {
if invocations.is_empty() {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked(
SkillInvokedInput {
tracking,
invocations,
},
)));
}
pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec<AppInvocation>) {
if mentions.is_empty() {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppMentioned(
AppMentionedInput { tracking, mentions },
)));
}
pub fn track_app_used(&self, tracking: TrackEventsContext, app: AppInvocation) {
if !self.queue.should_enqueue_app_used(&tracking, &app) {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(
AppUsedInput { tracking, app },
)));
}
pub fn track_plugin_used(&self, tracking: TrackEventsContext, plugin: PluginTelemetryMetadata) {
if !self.queue.should_enqueue_plugin_used(&tracking, &plugin) {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed(
PluginUsedInput { tracking, plugin },
)));
}
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Installed,
}),
));
}
pub fn track_plugin_uninstalled(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Uninstalled,
}),
));
}
pub fn track_plugin_enabled(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Enabled,
}),
));
}
pub fn track_plugin_disabled(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Disabled,
}),
));
}
fn record_fact(&self, input: AnalyticsFact) {
if self.analytics_enabled == Some(false) {
return;
}
self.queue.try_send(input);
}
}
const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256;
const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10);
const ANALYTICS_EVENT_DEDUPE_MAX_KEYS: usize = 4096;
#[derive(Serialize)]
struct TrackEventsRequest {
events: Vec<TrackEventRequest>,
}
#[derive(Serialize)]
#[serde(untagged)]
enum TrackEventRequest {
SkillInvocation(SkillInvocationEventRequest),
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
PluginUsed(CodexPluginUsedEventRequest),
PluginInstalled(CodexPluginEventRequest),
PluginUninstalled(CodexPluginEventRequest),
PluginEnabled(CodexPluginEventRequest),
PluginDisabled(CodexPluginEventRequest),
}
#[derive(Serialize)]
struct SkillInvocationEventRequest {
event_type: &'static str,
skill_id: String,
skill_name: String,
event_params: SkillInvocationEventParams,
}
#[derive(Serialize)]
struct SkillInvocationEventParams {
product_client_id: Option<String>,
skill_scope: Option<String>,
repo_url: Option<String>,
thread_id: Option<String>,
invoke_type: Option<InvocationType>,
model_slug: Option<String>,
}
#[derive(Serialize)]
struct CodexAppMetadata {
connector_id: Option<String>,
thread_id: Option<String>,
turn_id: Option<String>,
app_name: Option<String>,
product_client_id: Option<String>,
invoke_type: Option<InvocationType>,
model_slug: Option<String>,
}
#[derive(Serialize)]
struct CodexAppMentionedEventRequest {
event_type: &'static str,
event_params: CodexAppMetadata,
}
#[derive(Serialize)]
struct CodexAppUsedEventRequest {
event_type: &'static str,
event_params: CodexAppMetadata,
}
#[derive(Serialize)]
struct CodexPluginMetadata {
plugin_id: Option<String>,
plugin_name: Option<String>,
marketplace_name: Option<String>,
has_skills: Option<bool>,
mcp_server_count: Option<usize>,
connector_ids: Option<Vec<String>>,
product_client_id: Option<String>,
}
#[derive(Serialize)]
struct CodexPluginUsedMetadata {
#[serde(flatten)]
plugin: CodexPluginMetadata,
thread_id: Option<String>,
turn_id: Option<String>,
model_slug: Option<String>,
}
#[derive(Serialize)]
struct CodexPluginEventRequest {
event_type: &'static str,
event_params: CodexPluginMetadata,
}
#[derive(Serialize)]
struct CodexPluginUsedEventRequest {
event_type: &'static str,
event_params: CodexPluginUsedMetadata,
}
impl AnalyticsReducer {
async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec<TrackEventRequest>) {
match input {
AnalyticsFact::Initialize {
connection_id: _connection_id,
params: _params,
} => {}
AnalyticsFact::Request {
connection_id: _connection_id,
request_id: _request_id,
request: _request,
} => {}
AnalyticsFact::Response {
connection_id: _connection_id,
response: _response,
} => {}
AnalyticsFact::Notification(_notification) => {}
AnalyticsFact::Custom(input) => match input {
CustomAnalyticsFact::SkillInvoked(input) => {
self.ingest_skill_invoked(input, out).await;
}
CustomAnalyticsFact::AppMentioned(input) => {
self.ingest_app_mentioned(input, out);
}
CustomAnalyticsFact::AppUsed(input) => {
self.ingest_app_used(input, out);
}
CustomAnalyticsFact::PluginUsed(input) => {
self.ingest_plugin_used(input, out);
}
CustomAnalyticsFact::PluginStateChanged(input) => {
self.ingest_plugin_state_changed(input, out);
}
},
}
}
async fn ingest_skill_invoked(
&mut self,
input: SkillInvokedInput,
out: &mut Vec<TrackEventRequest>,
) {
let SkillInvokedInput {
tracking,
invocations,
} = input;
for invocation in invocations {
let skill_scope = match invocation.skill_scope {
SkillScope::User => "user",
SkillScope::Repo => "repo",
SkillScope::System => "system",
SkillScope::Admin => "admin",
};
let repo_root = get_git_repo_root(invocation.skill_path.as_path());
let repo_url = if let Some(root) = repo_root.as_ref() {
collect_git_info(root)
.await
.and_then(|info| info.repository_url)
} else {
None
};
let skill_id = skill_id_for_local_skill(
repo_url.as_deref(),
repo_root.as_deref(),
invocation.skill_path.as_path(),
invocation.skill_name.as_str(),
);
out.push(TrackEventRequest::SkillInvocation(
SkillInvocationEventRequest {
event_type: "skill_invocation",
skill_id,
skill_name: invocation.skill_name.clone(),
event_params: SkillInvocationEventParams {
thread_id: Some(tracking.thread_id.clone()),
invoke_type: Some(invocation.invocation_type),
model_slug: Some(tracking.model_slug.clone()),
product_client_id: Some(originator().value),
repo_url,
skill_scope: Some(skill_scope.to_string()),
},
},
));
}
}
fn ingest_app_mentioned(&mut self, input: AppMentionedInput, out: &mut Vec<TrackEventRequest>) {
let AppMentionedInput { tracking, mentions } = input;
out.extend(mentions.into_iter().map(|mention| {
let event_params = codex_app_metadata(&tracking, mention);
TrackEventRequest::AppMentioned(CodexAppMentionedEventRequest {
event_type: "codex_app_mentioned",
event_params,
})
}));
}
fn ingest_app_used(&mut self, input: AppUsedInput, out: &mut Vec<TrackEventRequest>) {
let AppUsedInput { tracking, app } = input;
let event_params = codex_app_metadata(&tracking, app);
out.push(TrackEventRequest::AppUsed(CodexAppUsedEventRequest {
event_type: "codex_app_used",
event_params,
}));
}
fn ingest_plugin_used(&mut self, input: PluginUsedInput, out: &mut Vec<TrackEventRequest>) {
let PluginUsedInput { tracking, plugin } = input;
out.push(TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest {
event_type: "codex_plugin_used",
event_params: codex_plugin_used_metadata(&tracking, plugin),
}));
}
fn ingest_plugin_state_changed(
&mut self,
input: PluginStateChangedInput,
out: &mut Vec<TrackEventRequest>,
) {
let PluginStateChangedInput { plugin, state } = input;
let event = CodexPluginEventRequest {
event_type: plugin_state_event_type(state),
event_params: codex_plugin_metadata(plugin),
};
out.push(match state {
PluginState::Installed => TrackEventRequest::PluginInstalled(event),
PluginState::Uninstalled => TrackEventRequest::PluginUninstalled(event),
PluginState::Enabled => TrackEventRequest::PluginEnabled(event),
PluginState::Disabled => TrackEventRequest::PluginDisabled(event),
});
}
}
fn plugin_state_event_type(state: PluginState) -> &'static str {
match state {
PluginState::Installed => "codex_plugin_installed",
PluginState::Uninstalled => "codex_plugin_uninstalled",
PluginState::Enabled => "codex_plugin_enabled",
PluginState::Disabled => "codex_plugin_disabled",
}
}
fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> CodexAppMetadata {
CodexAppMetadata {
connector_id: app.connector_id,
thread_id: Some(tracking.thread_id.clone()),
turn_id: Some(tracking.turn_id.clone()),
app_name: app.app_name,
product_client_id: Some(originator().value),
invoke_type: app.invocation_type,
model_slug: Some(tracking.model_slug.clone()),
}
}
fn codex_plugin_metadata(plugin: PluginTelemetryMetadata) -> CodexPluginMetadata {
let capability_summary = plugin.capability_summary;
CodexPluginMetadata {
plugin_id: Some(plugin.plugin_id.as_key()),
plugin_name: Some(plugin.plugin_id.plugin_name),
marketplace_name: Some(plugin.plugin_id.marketplace_name),
has_skills: capability_summary
.as_ref()
.map(|summary| summary.has_skills),
mcp_server_count: capability_summary
.as_ref()
.map(|summary| summary.mcp_server_names.len()),
connector_ids: capability_summary.map(|summary| {
summary
.app_connector_ids
.into_iter()
.map(|connector_id| connector_id.0)
.collect()
}),
product_client_id: Some(originator().value),
}
}
fn codex_plugin_used_metadata(
tracking: &TrackEventsContext,
plugin: PluginTelemetryMetadata,
) -> CodexPluginUsedMetadata {
CodexPluginUsedMetadata {
plugin: codex_plugin_metadata(plugin),
thread_id: Some(tracking.thread_id.clone()),
turn_id: Some(tracking.turn_id.clone()),
model_slug: Some(tracking.model_slug.clone()),
}
}
async fn send_track_events(
auth_manager: &AuthManager,
base_url: &str,
events: Vec<TrackEventRequest>,
) {
if events.is_empty() {
return;
}
let Some(auth) = auth_manager.auth().await else {
return;
};
if !auth.is_chatgpt_auth() {
return;
}
let access_token = match auth.get_token() {
Ok(token) => token,
Err(_) => return,
};
let Some(account_id) = auth.get_account_id() else {
return;
};
let base_url = base_url.trim_end_matches('/');
let url = format!("{base_url}/codex/analytics-events/events");
let payload = TrackEventsRequest { events };
let response = create_client()
.post(&url)
.timeout(ANALYTICS_EVENTS_TIMEOUT)
.bearer_auth(&access_token)
.header("chatgpt-account-id", &account_id)
.header("Content-Type", "application/json")
.json(&payload)
.send()
.await;
match response {
Ok(response) if response.status().is_success() => {}
Ok(response) => {
let status = response.status();
let body = response.text().await.unwrap_or_default();
tracing::warn!("events failed with status {status}: {body}");
}
Err(err) => {
tracing::warn!("failed to send events request: {err}");
}
}
}
pub(crate) fn skill_id_for_local_skill(
repo_url: Option<&str>,
repo_root: Option<&Path>,
skill_path: &Path,
skill_name: &str,
) -> String {
let path = normalize_path_for_skill_id(repo_url, repo_root, skill_path);
let prefix = if let Some(url) = repo_url {
format!("repo_{url}")
} else {
"personal".to_string()
};
let raw_id = format!("{prefix}_{path}_{skill_name}");
let mut hasher = Sha1::new();
hasher.update(raw_id.as_bytes());
format!("{:x}", hasher.finalize())
}
/// Returns a normalized path for skill ID construction.
///
/// - Repo-scoped skills use a path relative to the repo root.
/// - User/admin/system skills use an absolute path.
fn normalize_path_for_skill_id(
repo_url: Option<&str>,
repo_root: Option<&Path>,
skill_path: &Path,
) -> String {
let resolved_path =
std::fs::canonicalize(skill_path).unwrap_or_else(|_| skill_path.to_path_buf());
match (repo_url, repo_root) {
(Some(_), Some(root)) => {
let resolved_root = std::fs::canonicalize(root).unwrap_or_else(|_| root.to_path_buf());
resolved_path
.strip_prefix(&resolved_root)
.unwrap_or(resolved_path.as_path())
.to_string_lossy()
.replace('\\', "/")
}
_ => resolved_path.to_string_lossy().replace('\\', "/"),
}
}
#[cfg(test)]
#[path = "analytics_client_tests.rs"]
mod tests;
codex-rs/analytics/src/analytics_client_tests.rs
+268 -27
View File
@@ -1,26 +1,47 @@
use super::AnalyticsEventsQueue;
use super::AnalyticsFact;
use super::AnalyticsReducer;
use super::AppInvocation;
use super::AppMentionedInput;
use super::AppUsedInput;
use super::CodexAppMentionedEventRequest;
use super::CodexAppUsedEventRequest;
use super::CodexPluginEventRequest;
use super::CodexPluginUsedEventRequest;
use super::CustomAnalyticsFact;
use super::InvocationType;
use super::PluginState;
use super::PluginStateChangedInput;
use super::PluginUsedInput;
use super::SkillInvocation;
use super::SkillInvokedInput;
use super::TrackEventRequest;
use super::TrackEventsContext;
use super::codex_app_metadata;
use super::codex_plugin_metadata;
use super::codex_plugin_used_metadata;
use super::normalize_path_for_skill_id;
use crate::client::AnalyticsEventsQueue;
use crate::events::AppServerRpcTransport;
use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::ThreadInitializationMode;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_plugin_metadata;
use crate::events::codex_plugin_used_metadata;
use crate::facts::AnalyticsFact;
use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CustomAnalyticsFact;
use crate::facts::InvocationType;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::PluginUsedInput;
use crate::facts::SkillInvocation;
use crate::facts::SkillInvokedInput;
use crate::facts::TrackEventsContext;
use crate::reducer::AnalyticsReducer;
use crate::reducer::normalize_path_for_skill_id;
use crate::reducer::skill_id_for_local_skill;
use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer;
use codex_app_server_protocol::AskForApproval as AppServerAskForApproval;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy;
use codex_app_server_protocol::SessionSource as AppServerSessionSource;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus;
use codex_login::default_client::DEFAULT_ORIGINATOR;
use codex_login::default_client::originator;
use codex_plugin::AppConnectorId;
use codex_plugin::PluginCapabilitySummary;
@@ -34,6 +55,61 @@ use std::sync::Arc;
use std::sync::Mutex;
use tokio::sync::mpsc;
fn sample_thread(thread_id: &str, ephemeral: bool) -> Thread {
Thread {
id: thread_id.to_string(),
preview: "first prompt".to_string(),
ephemeral,
model_provider: "openai".to_string(),
created_at: 1,
updated_at: 2,
status: AppServerThreadStatus::Idle,
path: None,
cwd: PathBuf::from("/tmp"),
cli_version: "0.0.0".to_string(),
source: AppServerSessionSource::Exec,
agent_nickname: None,
agent_role: None,
git_info: None,
name: None,
turns: Vec::new(),
}
}
fn sample_thread_start_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse {
ClientResponse::ThreadStart {
request_id: RequestId::Integer(1),
response: ThreadStartResponse {
thread: sample_thread(thread_id, ephemeral),
model: model.to_string(),
model_provider: "openai".to_string(),
service_tier: None,
cwd: PathBuf::from("/tmp"),
approval_policy: AppServerAskForApproval::OnFailure,
approvals_reviewer: AppServerApprovalsReviewer::User,
sandbox: AppServerSandboxPolicy::DangerFullAccess,
reasoning_effort: None,
},
}
}
fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse {
ClientResponse::ThreadResume {
request_id: RequestId::Integer(2),
response: ThreadResumeResponse {
thread: sample_thread(thread_id, ephemeral),
model: model.to_string(),
model_provider: "openai".to_string(),
service_tier: None,
cwd: PathBuf::from("/tmp"),
approval_policy: AppServerAskForApproval::OnFailure,
approvals_reviewer: AppServerApprovalsReviewer::User,
sandbox: AppServerSandboxPolicy::DangerFullAccess,
reasoning_effort: None,
},
}
}
fn expected_absolute_path(path: &PathBuf) -> String {
std::fs::canonicalize(path)
.unwrap_or_else(|_| path.to_path_buf())
@@ -204,6 +280,171 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() {
assert_eq!(queue.should_enqueue_app_used(&turn_2, &app), true);
}
#[test]
fn thread_initialized_event_serializes_expected_shape() {
let event = TrackEventRequest::ThreadInitialized(ThreadInitializedEvent {
event_type: "codex_thread_initialized",
event_params: ThreadInitializedEventParams {
thread_id: "thread-0".to_string(),
app_server_client: CodexAppServerClientMetadata {
product_client_id: DEFAULT_ORIGINATOR.to_string(),
client_name: Some("codex-tui".to_string()),
client_version: Some("1.0.0".to_string()),
rpc_transport: AppServerRpcTransport::Stdio,
experimental_api_enabled: Some(true),
},
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.1.0".to_string(),
runtime_os: "macos".to_string(),
runtime_os_version: "15.3.1".to_string(),
runtime_arch: "aarch64".to_string(),
},
model: "gpt-5".to_string(),
ephemeral: true,
thread_source: Some("user"),
initialization_mode: ThreadInitializationMode::New,
subagent_source: None,
parent_thread_id: None,
created_at: 1,
},
});
let payload = serde_json::to_value(&event).expect("serialize thread initialized event");
assert_eq!(
payload,
json!({
"event_type": "codex_thread_initialized",
"event_params": {
"thread_id": "thread-0",
"app_server_client": {
"product_client_id": DEFAULT_ORIGINATOR,
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"model": "gpt-5",
"ephemeral": true,
"thread_source": "user",
"initialization_mode": "new",
"subagent_source": null,
"parent_thread_id": null,
"created_at": 1
}
})
);
}
#[tokio::test]
async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialized() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::Response {
connection_id: 7,
response: Box::new(sample_thread_start_response(
"thread-no-client",
/*ephemeral*/ false,
"gpt-5",
)),
},
&mut events,
)
.await;
assert!(events.is_empty(), "thread events should require initialize");
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: Some(InitializeCapabilities {
experimental_api: false,
opt_out_notification_methods: None,
}),
},
product_client_id: DEFAULT_ORIGINATOR.to_string(),
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.99.0".to_string(),
runtime_os: "linux".to_string(),
runtime_os_version: "24.04".to_string(),
runtime_arch: "x86_64".to_string(),
},
rpc_transport: AppServerRpcTransport::Websocket,
},
&mut events,
)
.await;
assert!(events.is_empty(), "initialize should not publish by itself");
reducer
.ingest(
AnalyticsFact::Response {
connection_id: 7,
response: Box::new(sample_thread_resume_response(
"thread-1", /*ephemeral*/ true, "gpt-5",
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_thread_initialized");
assert_eq!(
payload[0]["event_params"]["app_server_client"]["product_client_id"],
DEFAULT_ORIGINATOR
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_name"],
"codex-tui"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_version"],
"1.0.0"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["rpc_transport"],
"websocket"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["experimental_api_enabled"],
false
);
assert_eq!(
payload[0]["event_params"]["runtime"]["codex_rs_version"],
"0.99.0"
);
assert_eq!(payload[0]["event_params"]["runtime"]["runtime_os"], "linux");
assert_eq!(
payload[0]["event_params"]["runtime"]["runtime_os_version"],
"24.04"
);
assert_eq!(
payload[0]["event_params"]["runtime"]["runtime_arch"],
"x86_64"
);
assert_eq!(payload[0]["event_params"]["initialization_mode"], "resumed");
assert_eq!(payload[0]["event_params"]["thread_source"], "user");
assert_eq!(payload[0]["event_params"]["subagent_source"], json!(null));
assert_eq!(payload[0]["event_params"]["parent_thread_id"], json!(null));
}
#[test]
fn plugin_used_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
@@ -292,7 +533,7 @@ fn plugin_used_dedupe_is_keyed_by_turn_and_plugin() {
#[tokio::test]
async fn reducer_ingests_skill_invoked_fact() {
let mut reducer = AnalyticsReducer;
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
@@ -300,7 +541,7 @@ async fn reducer_ingests_skill_invoked_fact() {
turn_id: "turn-1".to_string(),
};
let skill_path = PathBuf::from("/Users/abc/.codex/skills/doc/SKILL.md");
let expected_skill_id = super::skill_id_for_local_skill(
let expected_skill_id = skill_id_for_local_skill(
/*repo_url*/ None,
/*repo_root*/ None,
skill_path.as_path(),
@@ -343,7 +584,7 @@ async fn reducer_ingests_skill_invoked_fact() {
#[tokio::test]
async fn reducer_ingests_app_and_plugin_facts() {
let mut reducer = AnalyticsReducer;
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
@@ -396,7 +637,7 @@ async fn reducer_ingests_app_and_plugin_facts() {
#[tokio::test]
async fn reducer_ingests_plugin_state_changed_fact() {
let mut reducer = AnalyticsReducer;
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
codex-rs/analytics/src/client.rs
+272
View File
@@ -0,0 +1,272 @@
use crate::events::AppServerRpcTransport;
use crate::events::TrackEventRequest;
use crate::events::TrackEventsRequest;
use crate::events::current_runtime_metadata;
use crate::facts::AnalyticsFact;
use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CustomAnalyticsFact;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::SkillInvocation;
use crate::facts::SkillInvokedInput;
use crate::facts::TrackEventsContext;
use crate::reducer::AnalyticsReducer;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::InitializeParams;
use codex_login::AuthManager;
use codex_login::default_client::create_client;
use codex_plugin::PluginTelemetryMetadata;
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use tokio::sync::mpsc;
const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256;
const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10);
const ANALYTICS_EVENT_DEDUPE_MAX_KEYS: usize = 4096;
#[derive(Clone)]
pub(crate) struct AnalyticsEventsQueue {
pub(crate) sender: mpsc::Sender<AnalyticsFact>,
pub(crate) app_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
pub(crate) plugin_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
}
#[derive(Clone)]
pub struct AnalyticsEventsClient {
queue: AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
}
impl AnalyticsEventsQueue {
pub(crate) fn new(auth_manager: Arc<AuthManager>, base_url: String) -> Self {
let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE);
tokio::spawn(async move {
let mut reducer = AnalyticsReducer::default();
while let Some(input) = receiver.recv().await {
let mut events = Vec::new();
reducer.ingest(input, &mut events).await;
send_track_events(&auth_manager, &base_url, events).await;
}
});
Self {
sender,
app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
}
}
fn try_send(&self, input: AnalyticsFact) {
if self.sender.try_send(input).is_err() {
//TODO: add a metric for this
tracing::warn!("dropping analytics events: queue is full");
}
}
pub(crate) fn should_enqueue_app_used(
&self,
tracking: &TrackEventsContext,
app: &AppInvocation,
) -> bool {
let Some(connector_id) = app.connector_id.as_ref() else {
return true;
};
let mut emitted = self
.app_used_emitted_keys
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if emitted.len() >= ANALYTICS_EVENT_DEDUPE_MAX_KEYS {
emitted.clear();
}
emitted.insert((tracking.turn_id.clone(), connector_id.clone()))
}
pub(crate) fn should_enqueue_plugin_used(
&self,
tracking: &TrackEventsContext,
plugin: &PluginTelemetryMetadata,
) -> bool {
let mut emitted = self
.plugin_used_emitted_keys
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if emitted.len() >= ANALYTICS_EVENT_DEDUPE_MAX_KEYS {
emitted.clear();
}
emitted.insert((tracking.turn_id.clone(), plugin.plugin_id.as_key()))
}
}
impl AnalyticsEventsClient {
pub fn new(
auth_manager: Arc<AuthManager>,
base_url: String,
analytics_enabled: Option<bool>,
) -> Self {
Self {
queue: AnalyticsEventsQueue::new(Arc::clone(&auth_manager), base_url),
analytics_enabled,
}
}
pub fn track_skill_invocations(
&self,
tracking: TrackEventsContext,
invocations: Vec<SkillInvocation>,
) {
if invocations.is_empty() {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked(
SkillInvokedInput {
tracking,
invocations,
},
)));
}
pub fn track_initialize(
&self,
connection_id: u64,
params: InitializeParams,
product_client_id: String,
rpc_transport: AppServerRpcTransport,
) {
self.record_fact(AnalyticsFact::Initialize {
connection_id,
params,
product_client_id,
runtime: current_runtime_metadata(),
rpc_transport,
});
}
pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec<AppInvocation>) {
if mentions.is_empty() {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppMentioned(
AppMentionedInput { tracking, mentions },
)));
}
pub fn track_app_used(&self, tracking: TrackEventsContext, app: AppInvocation) {
if !self.queue.should_enqueue_app_used(&tracking, &app) {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(
AppUsedInput { tracking, app },
)));
}
pub fn track_plugin_used(&self, tracking: TrackEventsContext, plugin: PluginTelemetryMetadata) {
if !self.queue.should_enqueue_plugin_used(&tracking, &plugin) {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed(
crate::facts::PluginUsedInput { tracking, plugin },
)));
}
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Installed,
}),
));
}
pub fn track_plugin_uninstalled(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Uninstalled,
}),
));
}
pub fn track_plugin_enabled(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Enabled,
}),
));
}
pub fn track_plugin_disabled(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Disabled,
}),
));
}
pub(crate) fn record_fact(&self, input: AnalyticsFact) {
if self.analytics_enabled == Some(false) {
return;
}
self.queue.try_send(input);
}
pub fn track_response(&self, connection_id: u64, response: ClientResponse) {
self.record_fact(AnalyticsFact::Response {
connection_id,
response: Box::new(response),
});
}
}
async fn send_track_events(
auth_manager: &AuthManager,
base_url: &str,
events: Vec<TrackEventRequest>,
) {
if events.is_empty() {
return;
}
let Some(auth) = auth_manager.auth().await else {
return;
};
if !auth.is_chatgpt_auth() {
return;
}
let access_token = match auth.get_token() {
Ok(token) => token,
Err(_) => return,
};
let Some(account_id) = auth.get_account_id() else {
return;
};
let base_url = base_url.trim_end_matches('/');
let url = format!("{base_url}/codex/analytics-events/events");
let payload = TrackEventsRequest { events };
let response = create_client()
.post(&url)
.timeout(ANALYTICS_EVENTS_TIMEOUT)
.bearer_auth(&access_token)
.header("chatgpt-account-id", &account_id)
.header("Content-Type", "application/json")
.json(&payload)
.send()
.await;
match response {
Ok(response) if response.status().is_success() => {}
Ok(response) => {
let status = response.status();
let body = response.text().await.unwrap_or_default();
tracing::warn!("events failed with status {status}: {body}");
}
Err(err) => {
tracing::warn!("failed to send events request: {err}");
}
}
}
codex-rs/analytics/src/events.rs
+230
View File
@@ -0,0 +1,230 @@
use crate::facts::AppInvocation;
use crate::facts::InvocationType;
use crate::facts::PluginState;
use crate::facts::TrackEventsContext;
use codex_login::default_client::originator;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::protocol::SessionSource;
use serde::Serialize;
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AppServerRpcTransport {
Stdio,
Websocket,
InProcess,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ThreadInitializationMode {
New,
Forked,
Resumed,
}
#[derive(Serialize)]
pub(crate) struct TrackEventsRequest {
pub(crate) events: Vec<TrackEventRequest>,
}
#[derive(Serialize)]
#[serde(untagged)]
pub(crate) enum TrackEventRequest {
SkillInvocation(SkillInvocationEventRequest),
ThreadInitialized(ThreadInitializedEvent),
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
PluginUsed(CodexPluginUsedEventRequest),
PluginInstalled(CodexPluginEventRequest),
PluginUninstalled(CodexPluginEventRequest),
PluginEnabled(CodexPluginEventRequest),
PluginDisabled(CodexPluginEventRequest),
}
#[derive(Serialize)]
pub(crate) struct SkillInvocationEventRequest {
pub(crate) event_type: &'static str,
pub(crate) skill_id: String,
pub(crate) skill_name: String,
pub(crate) event_params: SkillInvocationEventParams,
}
#[derive(Serialize)]
pub(crate) struct SkillInvocationEventParams {
pub(crate) product_client_id: Option<String>,
pub(crate) skill_scope: Option<String>,
pub(crate) repo_url: Option<String>,
pub(crate) thread_id: Option<String>,
pub(crate) invoke_type: Option<InvocationType>,
pub(crate) model_slug: Option<String>,
}
#[derive(Clone, Serialize)]
pub(crate) struct CodexAppServerClientMetadata {
pub(crate) product_client_id: String,
pub(crate) client_name: Option<String>,
pub(crate) client_version: Option<String>,
pub(crate) rpc_transport: AppServerRpcTransport,
pub(crate) experimental_api_enabled: Option<bool>,
}
#[derive(Clone, Serialize)]
pub(crate) struct CodexRuntimeMetadata {
pub(crate) codex_rs_version: String,
pub(crate) runtime_os: String,
pub(crate) runtime_os_version: String,
pub(crate) runtime_arch: String,
}
#[derive(Serialize)]
pub(crate) struct ThreadInitializedEventParams {
pub(crate) thread_id: String,
pub(crate) app_server_client: CodexAppServerClientMetadata,
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) model: String,
pub(crate) ephemeral: bool,
pub(crate) thread_source: Option<&'static str>,
pub(crate) initialization_mode: ThreadInitializationMode,
pub(crate) subagent_source: Option<String>,
pub(crate) parent_thread_id: Option<String>,
pub(crate) created_at: u64,
}
#[derive(Serialize)]
pub(crate) struct ThreadInitializedEvent {
pub(crate) event_type: &'static str,
pub(crate) event_params: ThreadInitializedEventParams,
}
#[derive(Serialize)]
pub(crate) struct CodexAppMetadata {
pub(crate) connector_id: Option<String>,
pub(crate) thread_id: Option<String>,
pub(crate) turn_id: Option<String>,
pub(crate) app_name: Option<String>,
pub(crate) product_client_id: Option<String>,
pub(crate) invoke_type: Option<InvocationType>,
pub(crate) model_slug: Option<String>,
}
#[derive(Serialize)]
pub(crate) struct CodexAppMentionedEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexAppMetadata,
}
#[derive(Serialize)]
pub(crate) struct CodexAppUsedEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexAppMetadata,
}
#[derive(Serialize)]
pub(crate) struct CodexPluginMetadata {
pub(crate) plugin_id: Option<String>,
pub(crate) plugin_name: Option<String>,
pub(crate) marketplace_name: Option<String>,
pub(crate) has_skills: Option<bool>,
pub(crate) mcp_server_count: Option<usize>,
pub(crate) connector_ids: Option<Vec<String>>,
pub(crate) product_client_id: Option<String>,
}
#[derive(Serialize)]
pub(crate) struct CodexPluginUsedMetadata {
#[serde(flatten)]
pub(crate) plugin: CodexPluginMetadata,
pub(crate) thread_id: Option<String>,
pub(crate) turn_id: Option<String>,
pub(crate) model_slug: Option<String>,
}
#[derive(Serialize)]
pub(crate) struct CodexPluginEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexPluginMetadata,
}
#[derive(Serialize)]
pub(crate) struct CodexPluginUsedEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexPluginUsedMetadata,
}
pub(crate) fn plugin_state_event_type(state: PluginState) -> &'static str {
match state {
PluginState::Installed => "codex_plugin_installed",
PluginState::Uninstalled => "codex_plugin_uninstalled",
PluginState::Enabled => "codex_plugin_enabled",
PluginState::Disabled => "codex_plugin_disabled",
}
}
pub(crate) fn codex_app_metadata(
tracking: &TrackEventsContext,
app: AppInvocation,
) -> CodexAppMetadata {
CodexAppMetadata {
connector_id: app.connector_id,
thread_id: Some(tracking.thread_id.clone()),
turn_id: Some(tracking.turn_id.clone()),
app_name: app.app_name,
product_client_id: Some(originator().value),
invoke_type: app.invocation_type,
model_slug: Some(tracking.model_slug.clone()),
}
}
pub(crate) fn codex_plugin_metadata(plugin: PluginTelemetryMetadata) -> CodexPluginMetadata {
let capability_summary = plugin.capability_summary;
CodexPluginMetadata {
plugin_id: Some(plugin.plugin_id.as_key()),
plugin_name: Some(plugin.plugin_id.plugin_name),
marketplace_name: Some(plugin.plugin_id.marketplace_name),
has_skills: capability_summary
.as_ref()
.map(|summary| summary.has_skills),
mcp_server_count: capability_summary
.as_ref()
.map(|summary| summary.mcp_server_names.len()),
connector_ids: capability_summary.map(|summary| {
summary
.app_connector_ids
.into_iter()
.map(|connector_id| connector_id.0)
.collect()
}),
product_client_id: Some(originator().value),
}
}
pub(crate) fn codex_plugin_used_metadata(
tracking: &TrackEventsContext,
plugin: PluginTelemetryMetadata,
) -> CodexPluginUsedMetadata {
CodexPluginUsedMetadata {
plugin: codex_plugin_metadata(plugin),
thread_id: Some(tracking.thread_id.clone()),
turn_id: Some(tracking.turn_id.clone()),
model_slug: Some(tracking.model_slug.clone()),
}
}
pub(crate) fn thread_source_name(thread_source: &SessionSource) -> Option<&'static str> {
match thread_source {
SessionSource::Cli | SessionSource::VSCode | SessionSource::Exec => Some("user"),
SessionSource::SubAgent(_) => Some("subagent"),
SessionSource::Mcp | SessionSource::Custom(_) | SessionSource::Unknown => None,
}
}
pub(crate) fn current_runtime_metadata() -> CodexRuntimeMetadata {
let os_info = os_info::get();
CodexRuntimeMetadata {
codex_rs_version: env!("CARGO_PKG_VERSION").to_string(),
runtime_os: std::env::consts::OS.to_string(),
runtime_os_version: os_info.version().to_string(),
runtime_arch: std::env::consts::ARCH.to_string(),
}
}
codex-rs/analytics/src/facts.rs
+116
View File
@@ -0,0 +1,116 @@
use crate::events::AppServerRpcTransport;
use crate::events::CodexRuntimeMetadata;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::protocol::SkillScope;
use serde::Serialize;
use std::path::PathBuf;
#[derive(Clone)]
pub struct TrackEventsContext {
pub model_slug: String,
pub thread_id: String,
pub turn_id: String,
}
pub fn build_track_events_context(
model_slug: String,
thread_id: String,
turn_id: String,
) -> TrackEventsContext {
TrackEventsContext {
model_slug,
thread_id,
turn_id,
}
}
#[derive(Clone, Debug)]
pub struct SkillInvocation {
pub skill_name: String,
pub skill_scope: SkillScope,
pub skill_path: PathBuf,
pub invocation_type: InvocationType,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum InvocationType {
Explicit,
Implicit,
}
pub struct AppInvocation {
pub connector_id: Option<String>,
pub app_name: Option<String>,
pub invocation_type: Option<InvocationType>,
}
#[allow(dead_code)]
pub(crate) enum AnalyticsFact {
Initialize {
connection_id: u64,
params: InitializeParams,
product_client_id: String,
runtime: CodexRuntimeMetadata,
rpc_transport: AppServerRpcTransport,
},
Request {
connection_id: u64,
request_id: RequestId,
request: Box<ClientRequest>,
},
Response {
connection_id: u64,
response: Box<ClientResponse>,
},
Notification(Box<ServerNotification>),
// Facts that do not naturally exist on the app-server protocol surface, or
// would require non-trivial protocol reshaping on this branch.
Custom(CustomAnalyticsFact),
}
pub(crate) enum CustomAnalyticsFact {
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
AppUsed(AppUsedInput),
PluginUsed(PluginUsedInput),
PluginStateChanged(PluginStateChangedInput),
}
pub(crate) struct SkillInvokedInput {
pub tracking: TrackEventsContext,
pub invocations: Vec<SkillInvocation>,
}
pub(crate) struct AppMentionedInput {
pub tracking: TrackEventsContext,
pub mentions: Vec<AppInvocation>,
}
pub(crate) struct AppUsedInput {
pub tracking: TrackEventsContext,
pub app: AppInvocation,
}
pub(crate) struct PluginUsedInput {
pub tracking: TrackEventsContext,
pub plugin: PluginTelemetryMetadata,
}
pub(crate) struct PluginStateChangedInput {
pub plugin: PluginTelemetryMetadata,
pub state: PluginState,
}
#[derive(Clone, Copy)]
pub(crate) enum PluginState {
Installed,
Uninstalled,
Enabled,
Disabled,
}
codex-rs/analytics/src/lib.rs
+14 -16
View File
@@ -1,17 +1,15 @@
mod analytics_client;
mod client;
mod events;
mod facts;
mod reducer;
pub use analytics_client::AnalyticsEventsClient;
pub use analytics_client::AnalyticsFact;
pub use analytics_client::AnalyticsReducer;
pub use analytics_client::AppInvocation;
pub use analytics_client::AppMentionedInput;
pub use analytics_client::AppUsedInput;
pub use analytics_client::CustomAnalyticsFact;
pub use analytics_client::InvocationType;
pub use analytics_client::PluginState;
pub use analytics_client::PluginStateChangedInput;
pub use analytics_client::PluginUsedInput;
pub use analytics_client::SkillInvocation;
pub use analytics_client::SkillInvokedInput;
pub use analytics_client::TrackEventsContext;
pub use analytics_client::build_track_events_context;
pub use client::AnalyticsEventsClient;
pub use events::AppServerRpcTransport;
pub use facts::AppInvocation;
pub use facts::InvocationType;
pub use facts::SkillInvocation;
pub use facts::TrackEventsContext;
pub use facts::build_track_events_context;
#[cfg(test)]
mod analytics_client_tests;
codex-rs/analytics/src/reducer.rs
+305
View File
@@ -0,0 +1,305 @@
use crate::events::AppServerRpcTransport;
use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::SkillInvocationEventParams;
use crate::events::SkillInvocationEventRequest;
use crate::events::ThreadInitializationMode;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_plugin_metadata;
use crate::events::codex_plugin_used_metadata;
use crate::events::plugin_state_event_type;
use crate::events::thread_source_name;
use crate::facts::AnalyticsFact;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CustomAnalyticsFact;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::PluginUsedInput;
use crate::facts::SkillInvokedInput;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::InitializeParams;
use codex_git_utils::collect_git_info;
use codex_git_utils::get_git_repo_root;
use codex_login::default_client::originator;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SkillScope;
use sha1::Digest;
use std::collections::HashMap;
use std::path::Path;
#[derive(Default)]
pub(crate) struct AnalyticsReducer {
connections: HashMap<u64, ConnectionState>,
}
struct ConnectionState {
app_server_client: CodexAppServerClientMetadata,
runtime: CodexRuntimeMetadata,
}
impl AnalyticsReducer {
pub(crate) async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec<TrackEventRequest>) {
match input {
AnalyticsFact::Initialize {
connection_id,
params,
product_client_id,
runtime,
rpc_transport,
} => {
self.ingest_initialize(
connection_id,
params,
product_client_id,
runtime,
rpc_transport,
);
}
AnalyticsFact::Request {
connection_id: _connection_id,
request_id: _request_id,
request: _request,
} => {}
AnalyticsFact::Response {
connection_id,
response,
} => {
self.ingest_response(connection_id, *response, out);
}
AnalyticsFact::Notification(_notification) => {}
AnalyticsFact::Custom(input) => match input {
CustomAnalyticsFact::SkillInvoked(input) => {
self.ingest_skill_invoked(input, out).await;
}
CustomAnalyticsFact::AppMentioned(input) => {
self.ingest_app_mentioned(input, out);
}
CustomAnalyticsFact::AppUsed(input) => {
self.ingest_app_used(input, out);
}
CustomAnalyticsFact::PluginUsed(input) => {
self.ingest_plugin_used(input, out);
}
CustomAnalyticsFact::PluginStateChanged(input) => {
self.ingest_plugin_state_changed(input, out);
}
},
}
}
fn ingest_initialize(
&mut self,
connection_id: u64,
params: InitializeParams,
product_client_id: String,
runtime: CodexRuntimeMetadata,
rpc_transport: AppServerRpcTransport,
) {
self.connections.insert(
connection_id,
ConnectionState {
app_server_client: CodexAppServerClientMetadata {
product_client_id,
client_name: Some(params.client_info.name),
client_version: Some(params.client_info.version),
rpc_transport,
experimental_api_enabled: params
.capabilities
.map(|capabilities| capabilities.experimental_api),
},
runtime,
},
);
}
async fn ingest_skill_invoked(
&mut self,
input: SkillInvokedInput,
out: &mut Vec<TrackEventRequest>,
) {
let SkillInvokedInput {
tracking,
invocations,
} = input;
for invocation in invocations {
let skill_scope = match invocation.skill_scope {
SkillScope::User => "user",
SkillScope::Repo => "repo",
SkillScope::System => "system",
SkillScope::Admin => "admin",
};
let repo_root = get_git_repo_root(invocation.skill_path.as_path());
let repo_url = if let Some(root) = repo_root.as_ref() {
collect_git_info(root)
.await
.and_then(|info| info.repository_url)
} else {
None
};
let skill_id = skill_id_for_local_skill(
repo_url.as_deref(),
repo_root.as_deref(),
invocation.skill_path.as_path(),
invocation.skill_name.as_str(),
);
out.push(TrackEventRequest::SkillInvocation(
SkillInvocationEventRequest {
event_type: "skill_invocation",
skill_id,
skill_name: invocation.skill_name.clone(),
event_params: SkillInvocationEventParams {
thread_id: Some(tracking.thread_id.clone()),
invoke_type: Some(invocation.invocation_type),
model_slug: Some(tracking.model_slug.clone()),
product_client_id: Some(originator().value),
repo_url,
skill_scope: Some(skill_scope.to_string()),
},
},
));
}
}
fn ingest_app_mentioned(&mut self, input: AppMentionedInput, out: &mut Vec<TrackEventRequest>) {
let AppMentionedInput { tracking, mentions } = input;
out.extend(mentions.into_iter().map(|mention| {
let event_params = codex_app_metadata(&tracking, mention);
TrackEventRequest::AppMentioned(CodexAppMentionedEventRequest {
event_type: "codex_app_mentioned",
event_params,
})
}));
}
fn ingest_app_used(&mut self, input: AppUsedInput, out: &mut Vec<TrackEventRequest>) {
let AppUsedInput { tracking, app } = input;
let event_params = codex_app_metadata(&tracking, app);
out.push(TrackEventRequest::AppUsed(CodexAppUsedEventRequest {
event_type: "codex_app_used",
event_params,
}));
}
fn ingest_plugin_used(&mut self, input: PluginUsedInput, out: &mut Vec<TrackEventRequest>) {
let PluginUsedInput { tracking, plugin } = input;
out.push(TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest {
event_type: "codex_plugin_used",
event_params: codex_plugin_used_metadata(&tracking, plugin),
}));
}
fn ingest_plugin_state_changed(
&mut self,
input: PluginStateChangedInput,
out: &mut Vec<TrackEventRequest>,
) {
let PluginStateChangedInput { plugin, state } = input;
let event = CodexPluginEventRequest {
event_type: plugin_state_event_type(state),
event_params: codex_plugin_metadata(plugin),
};
out.push(match state {
PluginState::Installed => TrackEventRequest::PluginInstalled(event),
PluginState::Uninstalled => TrackEventRequest::PluginUninstalled(event),
PluginState::Enabled => TrackEventRequest::PluginEnabled(event),
PluginState::Disabled => TrackEventRequest::PluginDisabled(event),
});
}
fn ingest_response(
&mut self,
connection_id: u64,
response: ClientResponse,
out: &mut Vec<TrackEventRequest>,
) {
let (thread, model, initialization_mode) = match response {
ClientResponse::ThreadStart { response, .. } => (
response.thread,
response.model,
ThreadInitializationMode::New,
),
ClientResponse::ThreadResume { response, .. } => (
response.thread,
response.model,
ThreadInitializationMode::Resumed,
),
ClientResponse::ThreadFork { response, .. } => (
response.thread,
response.model,
ThreadInitializationMode::Forked,
),
_ => return,
};
let thread_source: SessionSource = thread.source.into();
let Some(connection_state) = self.connections.get(&connection_id) else {
return;
};
out.push(TrackEventRequest::ThreadInitialized(
ThreadInitializedEvent {
event_type: "codex_thread_initialized",
event_params: ThreadInitializedEventParams {
thread_id: thread.id,
app_server_client: connection_state.app_server_client.clone(),
runtime: connection_state.runtime.clone(),
model,
ephemeral: thread.ephemeral,
thread_source: thread_source_name(&thread_source),
initialization_mode,
subagent_source: None,
parent_thread_id: None,
created_at: u64::try_from(thread.created_at).unwrap_or_default(),
},
},
));
}
}
pub(crate) fn skill_id_for_local_skill(
repo_url: Option<&str>,
repo_root: Option<&Path>,
skill_path: &Path,
skill_name: &str,
) -> String {
let path = normalize_path_for_skill_id(repo_url, repo_root, skill_path);
let prefix = if let Some(url) = repo_url {
format!("repo_{url}")
} else {
"personal".to_string()
};
let raw_id = format!("{prefix}_{path}_{skill_name}");
let mut hasher = sha1::Sha1::new();
sha1::Digest::update(&mut hasher, raw_id.as_bytes());
format!("{:x}", sha1::Digest::finalize(hasher))
}
/// Returns a normalized path for skill ID construction.
///
/// - Repo-scoped skills use a path relative to the repo root.
/// - User/admin/system skills use an absolute path.
pub(crate) fn normalize_path_for_skill_id(
repo_url: Option<&str>,
repo_root: Option<&Path>,
skill_path: &Path,
) -> String {
let resolved_path =
std::fs::canonicalize(skill_path).unwrap_or_else(|_| skill_path.to_path_buf());
match (repo_url, repo_root) {
(Some(_), Some(root)) => {
let resolved_root = std::fs::canonicalize(root).unwrap_or_else(|_| root.to_path_buf());
resolved_path
.strip_prefix(&resolved_root)
.unwrap_or(resolved_path.as_path())
.to_string_lossy()
.replace('\\', "/")
}
_ => resolved_path.to_string_lossy().replace('\\', "/"),
}
}
codex-rs/ansi-escape/Cargo.toml
+3
View File
@@ -8,6 +8,9 @@ license.workspace = true
name = "codex_ansi_escape"
path = "src/lib.rs"
[lints]
workspace = true
[dependencies]
ansi-to-tui = { workspace = true }
ratatui = { workspace = true, features = [
codex-rs/app-server/src/codex_message_processor.rs
+45
View File
@@ -32,6 +32,7 @@ use codex_app_server_protocol::CancelLoginAccountParams;
use codex_app_server_protocol::CancelLoginAccountResponse;
use codex_app_server_protocol::CancelLoginAccountStatus;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::CodexErrorInfo as AppServerCodexErrorInfo;
use codex_app_server_protocol::CollaborationModeListParams;
use codex_app_server_protocol::CollaborationModeListResponse;
@@ -186,6 +187,7 @@ use codex_arg0::Arg0DispatchPaths;
use codex_backend_client::Client as BackendClient;
use codex_chatgpt::connectors;
use codex_cloud_requirements::cloud_requirements_loader;
use codex_core::AnalyticsEventsClient;
use codex_core::AuthManager;
use codex_core::CodexAuth;
use codex_core::CodexThread;
@@ -410,6 +412,7 @@ pub(crate) struct CodexMessageProcessor {
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
arg0_paths: Arg0DispatchPaths,
config: Arc<Config>,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
@@ -440,6 +443,8 @@ struct ListenerTaskContext {
thread_manager: Arc<ThreadManager>,
thread_state_manager: ThreadStateManager,
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
general_analytics_enabled: bool,
thread_watch_manager: ThreadWatchManager,
fallback_model_provider: String,
codex_home: PathBuf,
@@ -462,6 +467,7 @@ pub(crate) struct CodexMessageProcessorArgs {
pub(crate) auth_manager: Arc<AuthManager>,
pub(crate) thread_manager: Arc<ThreadManager>,
pub(crate) outgoing: Arc<OutgoingMessageSender>,
pub(crate) analytics_events_client: AnalyticsEventsClient,
pub(crate) arg0_paths: Arg0DispatchPaths,
pub(crate) config: Arc<Config>,
pub(crate) cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
@@ -526,6 +532,7 @@ impl CodexMessageProcessor {
auth_manager,
thread_manager,
outgoing,
analytics_events_client,
arg0_paths,
config,
cli_overrides,
@@ -538,6 +545,7 @@ impl CodexMessageProcessor {
auth_manager,
thread_manager,
outgoing: outgoing.clone(),
analytics_events_client,
arg0_paths,
config,
cli_overrides,
@@ -2105,6 +2113,8 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -2337,6 +2347,17 @@ impl CodexMessageProcessor {
sandbox: config_snapshot.sandbox_policy.into(),
reasoning_effort: config_snapshot.reasoning_effort,
};
if listener_task_context.general_analytics_enabled {
listener_task_context
.analytics_events_client
.track_response(
request_id.connection_id.0,
ClientResponse::ThreadStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
listener_task_context
.outgoing
@@ -3905,6 +3926,15 @@ impl CodexMessageProcessor {
sandbox: session_configured.sandbox_policy.into(),
reasoning_effort: session_configured.reasoning_effort,
};
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadResume {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
self.outgoing.send_response(request_id, response).await;
}
@@ -4512,6 +4542,15 @@ impl CodexMessageProcessor {
sandbox: session_configured.sandbox_policy.into(),
reasoning_effort: session_configured.reasoning_effort,
};
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadFork {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
self.outgoing.send_response(request_id, response).await;
@@ -7093,6 +7132,8 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -7180,6 +7221,8 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -7211,6 +7254,8 @@ impl CodexMessageProcessor {
outgoing,
thread_manager,
thread_state_manager,
analytics_events_client: _,
general_analytics_enabled: _,
thread_watch_manager,
fallback_model_provider,
codex_home,
codex-rs/app-server/src/in_process.rs
+2
View File
@@ -74,6 +74,7 @@ use codex_app_server_protocol::Result;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_arg0::Arg0DispatchPaths;
use codex_core::AppServerRpcTransport;
use codex_core::config::Config;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
@@ -393,6 +394,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
config_warnings: args.config_warnings,
session_source: args.session_source,
enable_codex_api_key_env: args.enable_codex_api_key_env,
rpc_transport: AppServerRpcTransport::InProcess,
});
let mut thread_created_rx = processor.thread_created_receiver();
let mut session = ConnectionSessionState::default();
codex-rs/app-server/src/lib.rs
+9
View File
@@ -36,6 +36,7 @@ use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::TextPosition as AppTextPosition;
use codex_app_server_protocol::TextRange as AppTextRange;
use codex_core::AppServerRpcTransport;
use codex_core::ExecPolicyError;
use codex_core::check_execpolicy_for_warnings;
use codex_core::config_loader::ConfigLoadError;
@@ -623,6 +624,7 @@ pub async fn run_main_with_transport(
config_warnings,
session_source,
enable_codex_api_key_env: false,
rpc_transport: analytics_rpc_transport(transport),
});
let mut thread_created_rx = processor.thread_created_receiver();
let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count();
@@ -846,6 +848,13 @@ pub async fn run_main_with_transport(
Ok(())
}
fn analytics_rpc_transport(transport: AppServerTransport) -> AppServerRpcTransport {
match transport {
AppServerTransport::Stdio => AppServerRpcTransport::Stdio,
AppServerTransport::WebSocket { .. } => AppServerRpcTransport::Websocket,
}
}
#[cfg(test)]
mod tests {
use super::LogFormat;
codex-rs/app-server/src/message_processor.rs
+28 -6
View File
@@ -56,6 +56,7 @@ use codex_app_server_protocol::experimental_required_message;
use codex_arg0::Arg0DispatchPaths;
use codex_chatgpt::connectors;
use codex_core::AnalyticsEventsClient;
use codex_core::AppServerRpcTransport;
use codex_core::AuthManager;
use codex_core::ThreadManager;
use codex_core::config::Config;
@@ -70,9 +71,10 @@ use codex_core::models_manager::collaboration_mode_presets::CollaborationModesCo
use codex_exec_server::EnvironmentManager;
use codex_features::Feature;
use codex_feedback::CodexFeedback;
use codex_login::AuthMode as LoginAuthMode;
use codex_login::auth::ExternalAuth;
use codex_login::auth::ExternalAuthRefreshContext;
use codex_login::auth::ExternalAuthRefreshReason;
use codex_login::auth::ExternalAuthRefresher;
use codex_login::auth::ExternalAuthTokens;
use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionSource;
@@ -102,7 +104,11 @@ impl ExternalAuthRefreshBridge {
}
#[async_trait]
impl ExternalAuthRefresher for ExternalAuthRefreshBridge {
impl ExternalAuth for ExternalAuthRefreshBridge {
fn auth_mode(&self) -> LoginAuthMode {
LoginAuthMode::Chatgpt
}
async fn refresh(
&self,
context: ExternalAuthRefreshContext,
@@ -159,9 +165,11 @@ pub(crate) struct MessageProcessor {
external_agent_config_api: ExternalAgentConfigApi,
fs_api: FsApi,
auth_manager: Arc<AuthManager>,
analytics_events_client: AnalyticsEventsClient,
fs_watch_manager: FsWatchManager,
config: Arc<Config>,
config_warnings: Arc<Vec<ConfigWarningNotification>>,
rpc_transport: AppServerRpcTransport,
}
#[derive(Clone, Debug, Default)]
@@ -186,6 +194,7 @@ pub(crate) struct MessageProcessorArgs {
pub(crate) config_warnings: Vec<ConfigWarningNotification>,
pub(crate) session_source: SessionSource,
pub(crate) enable_codex_api_key_env: bool,
pub(crate) rpc_transport: AppServerRpcTransport,
}
impl MessageProcessor {
@@ -205,8 +214,9 @@ impl MessageProcessor {
config_warnings,
session_source,
enable_codex_api_key_env,
rpc_transport,
} = args;
let auth_manager = AuthManager::shared_with_external_chatgpt_auth_refresher(
let auth_manager = AuthManager::shared_with_external_auth(
config.codex_home.clone(),
enable_codex_api_key_env,
config.cli_auth_credentials_store_mode,
@@ -242,6 +252,7 @@ impl MessageProcessor {
auth_manager: auth_manager.clone(),
thread_manager: Arc::clone(&thread_manager),
outgoing: outgoing.clone(),
analytics_events_client: analytics_events_client.clone(),
arg0_paths,
config: Arc::clone(&config),
cli_overrides: cli_overrides.clone(),
@@ -262,7 +273,7 @@ impl MessageProcessor {
loader_overrides,
cloud_requirements,
thread_manager,
analytics_events_client,
analytics_events_client.clone(),
);
let external_agent_config_api = ExternalAgentConfigApi::new(config.codex_home.clone());
let fs_api = FsApi::default();
@@ -275,14 +286,16 @@ impl MessageProcessor {
external_agent_config_api,
fs_api,
auth_manager,
analytics_events_client,
fs_watch_manager,
config,
config_warnings: Arc::new(config_warnings),
rpc_transport,
}
}
pub(crate) fn clear_runtime_references(&self) {
self.auth_manager.clear_external_chatgpt_auth_refresher();
self.auth_manager.clear_external_auth();
}
pub(crate) async fn process_request(
@@ -547,6 +560,7 @@ impl MessageProcessor {
// shared thread when another connected client did not opt into
// experimental API). Proposed direction is instance-global first-write-wins
// with initialize-time mismatch rejection.
let analytics_initialize_params = params.clone();
let (experimental_api_enabled, opt_out_notification_methods) =
match params.capabilities {
Some(capabilities) => (
@@ -568,7 +582,7 @@ impl MessageProcessor {
session.app_server_client_name = Some(name.clone());
session.client_version = Some(version.clone());
let originator = name.clone();
if let Err(error) = set_default_originator(originator) {
if let Err(error) = set_default_originator(originator.clone()) {
match error {
SetOriginatorError::InvalidHeaderValue => {
let error = JSONRPCErrorError {
@@ -591,6 +605,14 @@ impl MessageProcessor {
}
}
}
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_initialize(
connection_id.0,
analytics_initialize_params,
originator,
self.rpc_transport,
);
}
set_default_client_residency_requirement(self.config.enforce_residency.value());
let user_agent_suffix = format!("{name}; {version}");
if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
codex-rs/app-server/src/message_processor/tracing_tests.rs
+2
View File
@@ -20,6 +20,7 @@ use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput;
use codex_arg0::Arg0DispatchPaths;
use codex_core::AppServerRpcTransport;
use codex_core::config::Config;
use codex_core::config::ConfigBuilder;
use codex_core::config_loader::CloudRequirementsLoader;
@@ -246,6 +247,7 @@ fn build_test_processor(
config_warnings: Vec::new(),
session_source: SessionSource::VSCode,
enable_codex_api_key_env: false,
rpc_transport: AppServerRpcTransport::Stdio,
});
(processor, outgoing_rx)
}
codex-rs/app-server/tests/common/Cargo.toml
+3
View File
@@ -7,6 +7,9 @@ license.workspace = true
[lib]
path = "lib.rs"
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
base64 = { workspace = true }
codex-rs/app-server/tests/common/models_cache.rs
+1 -1
View File
@@ -27,7 +27,7 @@ fn preset_to_info(preset: &ModelPreset, priority: i32) -> ModelInfo {
},
supported_in_api: preset.supported_in_api,
priority,
upgrade: preset.upgrade.as_ref().map(|u| u.into()),
upgrade: preset.upgrade.as_ref().map(Into::into),
base_instructions: "base instructions".to_string(),
model_messages: None,
supports_reasoning_summaries: false,
codex-rs/app-server/tests/suite/v2/analytics.rs
+101
View File
@@ -1,10 +1,23 @@
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::DEFAULT_CLIENT_NAME;
use app_test_support::write_chatgpt_auth;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::config::ConfigBuilder;
use codex_core::config::types::OtelExporterKind;
use codex_core::config::types::OtelHttpProtocol;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::collections::HashMap;
use std::path::Path;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::timeout;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
const SERVICE_VERSION: &str = "0.0.0-test";
@@ -65,3 +78,91 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> {
assert_eq!(has_metrics, true);
Ok(())
}
pub(crate) async fn enable_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> {
Mock::given(method("POST"))
.and(path("/codex/analytics-events/events"))
.respond_with(ResponseTemplate::new(200))
.mount(server)
.await;
write_chatgpt_auth(
codex_home,
ChatGptAuthFixture::new("chatgpt-token")
.account_id("account-123")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
Ok(())
}
pub(crate) async fn wait_for_analytics_payload(
server: &MockServer,
read_timeout: Duration,
) -> Result<Value> {
let body = timeout(read_timeout, async {
loop {
let Some(requests) = server.received_requests().await else {
tokio::time::sleep(Duration::from_millis(25)).await;
continue;
};
if let Some(request) = requests.iter().find(|request| {
request.method == "POST" && request.url.path() == "/codex/analytics-events/events"
}) {
break request.body.clone();
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await?;
serde_json::from_slice(&body).map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))
}
pub(crate) fn thread_initialized_event(payload: &Value) -> Result<&Value> {
let events = payload["events"]
.as_array()
.ok_or_else(|| anyhow::anyhow!("analytics payload missing events array"))?;
events
.iter()
.find(|event| event["event_type"] == "codex_thread_initialized")
.ok_or_else(|| anyhow::anyhow!("codex_thread_initialized event should be present"))
}
pub(crate) fn assert_basic_thread_initialized_event(
event: &Value,
thread_id: &str,
expected_model: &str,
initialization_mode: &str,
) {
assert_eq!(event["event_params"]["thread_id"], thread_id);
assert_eq!(
event["event_params"]["app_server_client"]["product_client_id"],
DEFAULT_CLIENT_NAME
);
assert_eq!(
event["event_params"]["app_server_client"]["client_name"],
DEFAULT_CLIENT_NAME
);
assert_eq!(
event["event_params"]["app_server_client"]["rpc_transport"],
"stdio"
);
assert_eq!(event["event_params"]["model"], expected_model);
assert_eq!(event["event_params"]["ephemeral"], false);
assert_eq!(event["event_params"]["thread_source"], "user");
assert_eq!(
event["event_params"]["subagent_source"],
serde_json::Value::Null
);
assert_eq!(
event["event_params"]["parent_thread_id"],
serde_json::Value::Null
);
assert_eq!(
event["event_params"]["initialization_mode"],
initialization_mode
);
assert!(event["event_params"]["created_at"].as_u64().is_some());
}
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+57
View File
@@ -38,6 +38,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
@@ -177,6 +182,50 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "forked");
Ok(())
}
#[tokio::test]
async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -249,6 +298,7 @@ async fn thread_fork_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -509,7 +559,13 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\n[features]\ngeneral_analytics = true\n".to_string()
} else {
String::new()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -521,6 +577,7 @@ sandbox_mode = "read-only"
chatgpt_base_url = "{chatgpt_base_url}"
model_provider = "mock_provider"
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+59
View File
@@ -70,6 +70,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
@@ -150,6 +155,51 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout_with_text_elements(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Vec::new(),
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
..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, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "gpt-5.2-codex", "resumed");
Ok(())
}
#[tokio::test]
async fn thread_resume_returns_rollout_history() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -1449,6 +1499,7 @@ async fn thread_resume_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -1827,6 +1878,7 @@ model_provider = "mock_provider"
[features]
personality = true
general_analytics = true
[model_providers.mock_provider]
name = "Mock provider for test"
@@ -1843,7 +1895,13 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &std::path::Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\ngeneral_analytics = true".to_string()
} else {
String::new()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -1858,6 +1916,7 @@ model_provider = "mock_provider"
[features]
personality = true
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
codex-rs/app-server/tests/suite/v2/thread_start.rs
+81
View File
@@ -26,6 +26,7 @@ use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::path::Path;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::timeout;
use wiremock::Mock;
@@ -34,6 +35,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
@@ -153,6 +159,73 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let req_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
assert_eq!(payload["events"].as_array().expect("events array").len(), 1);
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "new");
Ok(())
}
#[tokio::test]
async fn thread_start_does_not_track_thread_initialized_analytics_without_feature() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ false,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let req_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
)
.await??;
let _ = to_response::<ThreadStartResponse>(resp)?;
let payload = wait_for_analytics_payload(&server, Duration::from_millis(250)).await;
assert!(
payload.is_err(),
"thread analytics should be gated off when general_analytics is disabled"
);
Ok(())
}
#[tokio::test]
async fn thread_start_respects_project_config_from_cwd() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -455,6 +528,7 @@ async fn thread_start_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -538,7 +612,13 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\n[features]\ngeneral_analytics = true\n".to_string()
} else {
String::new()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -550,6 +630,7 @@ sandbox_mode = "read-only"
chatgpt_base_url = "{chatgpt_base_url}"
model_provider = "mock_provider"
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
codex-rs/backend-client/Cargo.toml
+3
View File
@@ -8,6 +8,9 @@ publish = false
[lib]
path = "src/lib.rs"
[lints]
workspace = true
[dependencies]
anyhow = "1"
serde = { version = "1", features = ["derive"] }
codex-rs/codex-backend-openapi-models/Cargo.toml
+3
View File
@@ -8,6 +8,9 @@ license.workspace = true
name = "codex_backend_openapi_models"
path = "src/lib.rs"
[lints]
workspace = true
# Important: generated code often violates our workspace lints.
# Allow unwrap/expect in this crate so the workspace builds cleanly
# after models are regenerated.
codex-rs/core/config.schema.json
+6
View File
@@ -392,6 +392,9 @@
"fast_mode": {
"type": "boolean"
},
"general_analytics": {
"type": "boolean"
},
"guardian_approval": {
"type": "boolean"
},
@@ -2089,6 +2092,9 @@
"fast_mode": {
"type": "boolean"
},
"general_analytics": {
"type": "boolean"
},
"guardian_approval": {
"type": "boolean"
},
codex-rs/core/src/lib.rs
+1
View File
@@ -136,6 +136,7 @@ pub type CodexConversation = CodexThread;
pub use auth::AuthManager;
pub use auth::CodexAuth;
pub use codex_analytics::AnalyticsEventsClient;
pub use codex_analytics::AppServerRpcTransport;
mod default_client_forwarding;
/// Default Codex HTTP client headers and reqwest construction.
codex-rs/core/src/models_manager/manager_tests.rs
+1 -1
View File
@@ -145,7 +145,7 @@ mv tokens.next tokens.txt
let script_path = tempdir.path().join("print-token.ps1");
std::fs::write(
&script_path,
r#"$lines = Get-Content -Path tokens.txt
r#"$lines = @(Get-Content -Path tokens.txt)
if ($lines.Count -eq 0) { exit 1 }
Write-Output $lines[0]
$lines | Select-Object -Skip 1 | Set-Content -Path tokens.txt
codex-rs/core/tests/common/Cargo.toml
+3
View File
@@ -7,6 +7,9 @@ license.workspace = true
[lib]
path = "lib.rs"
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
assert_cmd = { workspace = true }
codex-rs/core/tests/suite/client.rs
+1 -1
View File
@@ -190,7 +190,7 @@ mv tokens.next tokens.txt
let script_path = tempdir.path().join("print-token.ps1");
std::fs::write(
&script_path,
r#"$lines = Get-Content -Path tokens.txt
r#"$lines = @(Get-Content -Path tokens.txt)
if ($lines.Count -eq 0) { exit 1 }
Write-Output $lines[0]
$lines | Select-Object -Skip 1 | Set-Content -Path tokens.txt
codex-rs/core/tests/suite/plugins.rs
+14 -8
View File
@@ -365,13 +365,22 @@ async fn explicit_plugin_mentions_track_plugin_used_analytics() -> Result<()> {
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let deadline = Instant::now() + Duration::from_secs(10);
let analytics_request = loop {
let plugin_event = loop {
let requests = server.received_requests().await.unwrap_or_default();
if let Some(request) = requests
if let Some(event) = requests
.into_iter()
.find(|request| request.url.path() == "/codex/analytics-events/events")
.filter(|request| request.url.path() == "/codex/analytics-events/events")
.find_map(|request| {
let payload: serde_json::Value = serde_json::from_slice(&request.body).ok()?;
payload["events"].as_array().and_then(|events| {
events
.iter()
.find(|event| event["event_type"] == "codex_plugin_used")
.cloned()
})
})
{
break request;
break event;
}
if Instant::now() >= deadline {
panic!("timed out waiting for plugin analytics request");
@@ -379,10 +388,7 @@ async fn explicit_plugin_mentions_track_plugin_used_analytics() -> Result<()> {
tokio::time::sleep(Duration::from_millis(50)).await;
};
let payload: serde_json::Value =
serde_json::from_slice(&analytics_request.body).expect("analytics payload");
let event = &payload["events"][0];
assert_eq!(event["event_type"], "codex_plugin_used");
let event = plugin_event;
assert_eq!(event["event_params"]["plugin_id"], "sample@test");
assert_eq!(event["event_params"]["plugin_name"], "sample");
assert_eq!(event["event_params"]["marketplace_name"], "test");
codex-rs/debug-client/Cargo.toml
+3
View File
@@ -4,6 +4,9 @@ version.workspace = true
edition.workspace = true
license.workspace = true
[lints]
workspace = true
[dependencies]
anyhow.workspace = true
clap = { workspace = true, features = ["derive"] }
codex-rs/debug-client/src/client.rs
+1
View File
@@ -1,3 +1,4 @@
#![allow(clippy::expect_used)]
use std::io::BufRead;
use std::io::BufReader;
use std::io::Write;
codex-rs/debug-client/src/output.rs
+1
View File
@@ -1,3 +1,4 @@
#![allow(clippy::expect_used)]
use std::io;
use std::io::IsTerminal;
use std::io::Write;
codex-rs/debug-client/src/reader.rs
+2 -1
View File
@@ -1,3 +1,4 @@
#![allow(clippy::expect_used)]
use std::io::BufRead;
use std::io::BufReader;
use std::process::ChildStdout;
@@ -113,7 +114,7 @@ fn handle_server_request(
stdin: &Arc<Mutex<Option<std::process::ChildStdin>>>,
output: &Output,
) -> anyhow::Result<()> {
let server_request = match ServerRequest::try_from(request.clone()) {
let server_request = match ServerRequest::try_from(request) {
Ok(server_request) => server_request,
Err(_) => return Ok(()),
};
codex-rs/exec/src/lib.rs
+1 -1
View File
@@ -440,7 +440,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
config_warnings,
session_source: SessionSource::Exec,
enable_codex_api_key_env: true,
client_name: "codex-exec".to_string(),
client_name: "codex_exec".to_string(),
client_version: env!("CARGO_PKG_VERSION").to_string(),
experimental_api: true,
opt_out_notification_methods: Vec::new(),
codex-rs/features/src/lib.rs
+8
View File
@@ -124,6 +124,8 @@ pub enum Feature {
CodexGitCommit,
/// Enable runtime metrics snapshots via a manual reader.
RuntimeMetrics,
/// Enable thread lifecycle analytics emitted via the app-server analytics pipeline.
GeneralAnalytics,
/// Persist rollout metadata to a local SQLite database.
Sqlite,
/// Enable startup memory extraction and file-backed memory consolidation.
@@ -611,6 +613,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::GeneralAnalytics,
key: "general_analytics",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::Sqlite,
key: "sqlite",
codex-rs/features/src/tests.rs
+6
View File
@@ -126,6 +126,12 @@ fn tool_search_is_under_development_and_disabled_by_default() {
assert_eq!(Feature::ToolSearch.default_enabled(), false);
}
#[test]
fn general_analytics_is_under_development_and_disabled_by_default() {
assert_eq!(Feature::GeneralAnalytics.stage(), Stage::UnderDevelopment);
assert_eq!(Feature::GeneralAnalytics.default_enabled(), false);
}
#[test]
fn use_linux_sandbox_bwrap_is_a_removed_feature_key() {
assert_eq!(
codex-rs/feedback/Cargo.toml
+3
View File
@@ -4,6 +4,9 @@ version.workspace = true
edition.workspace = true
license.workspace = true
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
codex-protocol = { workspace = true }
codex-rs/feedback/src/lib.rs
+4 -1
View File
@@ -95,10 +95,12 @@ impl CodexFeedback {
pub fn snapshot(&self, session_id: Option<ThreadId>) -> FeedbackSnapshot {
let bytes = {
#[allow(clippy::expect_used)]
let guard = self.inner.ring.lock().expect("mutex poisoned");
guard.snapshot_bytes()
};
let tags = {
#[allow(clippy::expect_used)]
let guard = self.inner.tags.lock().expect("mutex poisoned");
guard.clone()
};
@@ -324,7 +326,7 @@ impl FeedbackSnapshot {
use sentry::protocol::Values;
event.exception = Values::from(vec![Exception {
ty: title.clone(),
ty: title,
value: Some(r.to_string()),
..Default::default()
}]);
@@ -430,6 +432,7 @@ where
return;
}
#[allow(clippy::expect_used)]
let mut guard = self.inner.tags.lock().expect("mutex poisoned");
for (key, value) in visitor.tags {
if guard.len() >= MAX_FEEDBACK_TAGS && !guard.contains_key(&key) {
codex-rs/file-search/Cargo.toml
+3
View File
@@ -12,6 +12,9 @@ path = "src/main.rs"
name = "codex_file_search"
path = "src/lib.rs"
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
clap = { workspace = true, features = ["derive"] }
codex-rs/file-search/src/lib.rs
+10 -4
View File
@@ -195,10 +195,10 @@ pub fn create_session(
threads: threads.get(),
compute_indices,
respect_gitignore,
cancelled: cancelled.clone(),
cancelled,
shutdown: Arc::new(AtomicBool::new(false)),
reporter,
work_tx: work_tx.clone(),
work_tx,
});
let matcher_inner = inner.clone();
@@ -611,13 +611,14 @@ struct RunReporter {
impl SessionReporter for RunReporter {
fn on_update(&self, snapshot: &FileSearchSnapshot) {
#[expect(clippy::unwrap_used)]
#[allow(clippy::unwrap_used)]
let mut guard = self.snapshot.write().unwrap();
*guard = snapshot.clone();
}
fn on_complete(&self) {
let (cv, mutex) = &self.completed;
#[allow(clippy::unwrap_used)]
let mut completed = mutex.lock().unwrap();
*completed = true;
cv.notify_all();
@@ -627,10 +628,15 @@ impl SessionReporter for RunReporter {
impl RunReporter {
fn wait_for_complete(&self) -> FileSearchSnapshot {
let (cv, mutex) = &self.completed;
#[allow(clippy::unwrap_used)]
let mut completed = mutex.lock().unwrap();
while !*completed {
completed = cv.wait(completed).unwrap();
#[allow(clippy::unwrap_used)]
{
completed = cv.wait(completed).unwrap();
}
}
#[allow(clippy::unwrap_used)]
self.snapshot.read().unwrap().clone()
}
}
codex-rs/file-search/src/main.rs
+7 -2
View File
@@ -27,8 +27,11 @@ struct StdioReporter {
impl Reporter for StdioReporter {
fn report_match(&self, file_match: &FileMatch) {
if self.write_output_as_json {
println!("{}", serde_json::to_string(&file_match).unwrap());
#[allow(clippy::unwrap_used)]
let json = serde_json::to_string(file_match).unwrap();
println!("{json}");
} else if self.show_indices {
#[allow(clippy::expect_used)]
let indices = file_match
.indices
.as_ref()
@@ -61,7 +64,9 @@ impl Reporter for StdioReporter {
fn warn_matches_truncated(&self, total_match_count: usize, shown_match_count: usize) {
if self.write_output_as_json {
let value = json!({"matches_truncated": true});
println!("{}", serde_json::to_string(&value).unwrap());
#[allow(clippy::unwrap_used)]
let json = serde_json::to_string(&value).unwrap();
println!("{json}");
} else {
eprintln!(
"Warning: showing {shown_match_count} out of {total_match_count} results. Provide a more specific pattern or increase the --limit.",
codex-rs/login/src/auth/auth_tests.rs
+40 -36
View File
@@ -283,6 +283,8 @@ async fn external_bearer_only_auth_manager_uses_cached_provider_token() {
assert_eq!(first.as_deref(), Some("provider-token"));
assert_eq!(second.as_deref(), Some("provider-token"));
assert_eq!(manager.auth_mode(), Some(crate::AuthMode::ApiKey));
assert_eq!(manager.get_api_auth_mode(), Some(ApiAuthMode::ApiKey));
}
#[tokio::test]
@@ -364,7 +366,7 @@ mv tokens.next tokens.txt
let script_path = tempdir.path().join("print-token.ps1");
std::fs::write(
&script_path,
r#"$lines = Get-Content -Path tokens.txt
r#"$lines = @(Get-Content -Path tokens.txt)
if ($lines.Count -eq 0) { exit 1 }
Write-Output $lines[0]
$lines | Select-Object -Skip 1 | Set-Content -Path tokens.txt
@@ -448,42 +450,8 @@ struct AuthFileParams {
}
fn write_auth_file(params: AuthFileParams, codex_home: &Path) -> std::io::Result<String> {
let fake_jwt = fake_jwt_for_auth_file_params(&params)?;
let auth_file = get_auth_file(codex_home);
// Create a minimal valid JWT for the id_token field.
#[derive(Serialize)]
struct Header {
alg: &'static str,
typ: &'static str,
}
let header = Header {
alg: "none",
typ: "JWT",
};
let mut auth_payload = serde_json::json!({
"chatgpt_user_id": "user-12345",
"user_id": "user-12345",
});
if let Some(chatgpt_plan_type) = params.chatgpt_plan_type {
auth_payload["chatgpt_plan_type"] = serde_json::Value::String(chatgpt_plan_type);
}
if let Some(chatgpt_account_id) = params.chatgpt_account_id {
let org_value = serde_json::Value::String(chatgpt_account_id);
auth_payload["chatgpt_account_id"] = org_value;
}
let payload = serde_json::json!({
"email": "user@example.com",
"email_verified": true,
"https://api.openai.com/auth": auth_payload,
});
let b64 = |b: &[u8]| base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b);
let header_b64 = b64(&serde_json::to_vec(&header)?);
let payload_b64 = b64(&serde_json::to_vec(&payload)?);
let signature_b64 = b64(b"sig");
let fake_jwt = format!("{header_b64}.{payload_b64}.{signature_b64}");
let auth_json_data = json!({
"OPENAI_API_KEY": params.openai_api_key,
"tokens": {
@@ -498,6 +466,42 @@ fn write_auth_file(params: AuthFileParams, codex_home: &Path) -> std::io::Result
Ok(fake_jwt)
}
fn fake_jwt_for_auth_file_params(params: &AuthFileParams) -> std::io::Result<String> {
#[derive(Serialize)]
struct Header {
alg: &'static str,
typ: &'static str,
}
let header = Header {
alg: "none",
typ: "JWT",
};
let mut auth_payload = serde_json::json!({
"chatgpt_user_id": "user-12345",
"user_id": "user-12345",
});
if let Some(chatgpt_plan_type) = params.chatgpt_plan_type.as_ref() {
auth_payload["chatgpt_plan_type"] = serde_json::Value::String(chatgpt_plan_type.clone());
}
if let Some(chatgpt_account_id) = params.chatgpt_account_id.as_ref() {
auth_payload["chatgpt_account_id"] = serde_json::Value::String(chatgpt_account_id.clone());
}
let payload = serde_json::json!({
"email": "user@example.com",
"email_verified": true,
"https://api.openai.com/auth": auth_payload,
});
let b64 = |b: &[u8]| base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b);
let header_b64 = b64(&serde_json::to_vec(&header)?);
let payload_b64 = b64(&serde_json::to_vec(&payload)?);
let signature_b64 = b64(b"sig");
Ok(format!("{header_b64}.{payload_b64}.{signature_b64}"))
}
async fn build_config(
codex_home: &Path,
forced_login_method: Option<ForcedLoginMethod>,
codex-rs/login/src/auth/external_bearer.rs
+40 -22
View File
@@ -1,3 +1,7 @@
use super::manager::ExternalAuth;
use super::manager::ExternalAuthRefreshContext;
use super::manager::ExternalAuthTokens;
use async_trait::async_trait;
use codex_protocol::config_types::ModelProviderAuthInfo;
use std::fmt;
use std::io;
@@ -10,47 +14,61 @@ use tokio::process::Command;
use tokio::sync::Mutex;
#[derive(Clone)]
pub(crate) struct ExternalBearerAuth {
pub(crate) struct BearerTokenRefresher {
state: Arc<ExternalBearerAuthState>,
}
impl ExternalBearerAuth {
impl BearerTokenRefresher {
pub(crate) fn new(config: ModelProviderAuthInfo) -> Self {
Self {
state: Arc::new(ExternalBearerAuthState::new(config)),
}
}
}
pub(crate) async fn resolve_access_token(&self) -> io::Result<String> {
let mut cached = self.state.cached_token.lock().await;
if let Some(cached_token) = cached.as_ref()
&& cached_token.fetched_at.elapsed() < self.state.config.refresh_interval()
{
return Ok(cached_token.access_token.clone());
}
#[async_trait]
impl ExternalAuth for BearerTokenRefresher {
fn auth_mode(&self) -> crate::AuthMode {
crate::AuthMode::ApiKey
}
async fn resolve(&self) -> io::Result<Option<ExternalAuthTokens>> {
let access_token = {
let mut cached = self.state.cached_token.lock().await;
if let Some(cached_token) = cached.as_ref()
&& cached_token.fetched_at.elapsed() < self.state.config.refresh_interval()
{
cached_token.access_token.clone()
} else {
let access_token = run_provider_auth_command(&self.state.config).await?;
*cached = Some(CachedExternalBearerToken {
access_token: access_token.clone(),
fetched_at: Instant::now(),
});
access_token
}
};
Ok(Some(ExternalAuthTokens::access_token_only(access_token)))
}
async fn refresh(
&self,
_context: ExternalAuthRefreshContext,
) -> io::Result<ExternalAuthTokens> {
let access_token = run_provider_auth_command(&self.state.config).await?;
let mut cached = self.state.cached_token.lock().await;
*cached = Some(CachedExternalBearerToken {
access_token: access_token.clone(),
fetched_at: Instant::now(),
});
Ok(access_token)
}
pub(crate) async fn refresh_after_unauthorized(&self) -> io::Result<()> {
let access_token = run_provider_auth_command(&self.state.config).await?;
let mut cached = self.state.cached_token.lock().await;
*cached = Some(CachedExternalBearerToken {
access_token,
fetched_at: Instant::now(),
});
Ok(())
Ok(ExternalAuthTokens::access_token_only(access_token))
}
}
impl fmt::Debug for ExternalBearerAuth {
impl fmt::Debug for BearerTokenRefresher {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ExternalBearerAuth").finish_non_exhaustive()
f.debug_struct("BearerTokenRefresher")
.finish_non_exhaustive()
}
}
codex-rs/login/src/auth/manager.rs
+82 -74
View File
@@ -18,7 +18,7 @@ use codex_app_server_protocol::AuthMode as ApiAuthMode;
use codex_protocol::config_types::ForcedLoginMethod;
use codex_protocol::config_types::ModelProviderAuthInfo;
use super::external_bearer::ExternalBearerAuth;
use super::external_bearer::BearerTokenRefresher;
use crate::auth::error::RefreshTokenFailedError;
use crate::auth::error::RefreshTokenFailedReason;
pub use crate::auth::storage::AuthCredentialsStoreMode;
@@ -143,11 +143,21 @@ pub struct ExternalAuthRefreshContext {
}
#[async_trait]
pub trait ExternalAuthRefresher: Send + Sync {
/// Pluggable auth provider used by `AuthManager` for externally managed auth flows.
///
/// Implementations may either resolve auth eagerly via `resolve()` or provide refreshed
/// credentials on demand via `refresh()`.
pub trait ExternalAuth: Send + Sync {
/// Indicates which top-level auth mode this external provider supplies.
fn auth_mode(&self) -> crate::AuthMode;
/// Returns cached or immediately available auth, if this provider can resolve it synchronously
/// from the caller's perspective.
async fn resolve(&self) -> std::io::Result<Option<ExternalAuthTokens>> {
Ok(None)
}
/// Refreshes auth in response to a manager-driven refresh attempt.
async fn refresh(
&self,
context: ExternalAuthRefreshContext,
@@ -853,27 +863,6 @@ struct AuthScopedRefreshFailure {
error: RefreshTokenFailedError,
}
#[derive(Clone)]
enum ExternalAuth {
Bearer(ExternalBearerAuth),
ChatgptRefresher(Arc<dyn ExternalAuthRefresher>),
}
impl Debug for ExternalAuth {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Bearer(_) => f
.debug_tuple("ExternalAuth::Bearer")
.field(&"present")
.finish(),
Self::ChatgptRefresher(_) => f
.debug_tuple("ExternalAuth::ChatgptRefresher")
.field(&"present")
.finish(),
}
}
}
impl Debug for CachedAuth {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CachedAuth")
@@ -928,7 +917,7 @@ enum UnauthorizedRecoveryMode {
//
// - External ChatGPT auth tokens (`chatgptAuthTokens`) are refreshed by asking
// the parent app for new tokens through the configured
// `ExternalAuthRefresher`, persisting them in the ephemeral auth store, and
// `ExternalAuth`, persisting them in the ephemeral auth store, and
// reloading the cached auth snapshot.
// - External bearer auth sources for custom model providers rerun the provider
// auth command without touching disk.
@@ -954,7 +943,7 @@ impl UnauthorizedRecovery {
fn new(manager: Arc<AuthManager>) -> Self {
let cached_auth = manager.auth_cached();
let expected_account_id = cached_auth.as_ref().and_then(CodexAuth::get_account_id);
let mode = if manager.has_external_bearer_auth()
let mode = if manager.has_external_api_key_auth()
|| cached_auth
.as_ref()
.is_some_and(CodexAuth::is_external_chatgpt_tokens)
@@ -976,7 +965,7 @@ impl UnauthorizedRecovery {
}
pub fn has_next(&self) -> bool {
if self.manager.has_external_bearer_auth() {
if self.manager.has_external_api_key_auth() {
return !matches!(self.step, UnauthorizedRecoveryStep::Done);
}
@@ -989,9 +978,7 @@ impl UnauthorizedRecovery {
return false;
}
if self.mode == UnauthorizedRecoveryMode::External
&& !self.manager.has_external_chatgpt_auth_refresher()
{
if self.mode == UnauthorizedRecoveryMode::External && !self.manager.has_external_auth() {
return false;
}
@@ -999,7 +986,7 @@ impl UnauthorizedRecovery {
}
pub fn unavailable_reason(&self) -> &'static str {
if self.manager.has_external_bearer_auth() {
if self.manager.has_external_api_key_auth() {
return if matches!(self.step, UnauthorizedRecoveryStep::Done) {
"recovery_exhausted"
} else {
@@ -1016,10 +1003,8 @@ impl UnauthorizedRecovery {
return "not_chatgpt_auth";
}
if self.mode == UnauthorizedRecoveryMode::External
&& !self.manager.has_external_chatgpt_auth_refresher()
{
return "no_external_refresher";
if self.mode == UnauthorizedRecoveryMode::External && !self.manager.has_external_auth() {
return "no_external_auth";
}
if matches!(self.step, UnauthorizedRecoveryStep::Done) {
@@ -1112,7 +1097,6 @@ impl UnauthorizedRecovery {
/// External modifications to `auth.json` will NOT be observed until
/// `reload()` is called explicitly. This matches the design goal of avoiding
/// different parts of the program seeing inconsistent auth data mid‑run.
#[derive(Debug)]
pub struct AuthManager {
codex_home: PathBuf,
inner: RwLock<CachedAuth>,
@@ -1120,7 +1104,26 @@ pub struct AuthManager {
auth_credentials_store_mode: AuthCredentialsStoreMode,
forced_chatgpt_workspace_id: RwLock<Option<String>>,
refresh_lock: AsyncMutex<()>,
external_auth: RwLock<Option<ExternalAuth>>,
external_auth: RwLock<Option<Arc<dyn ExternalAuth>>>,
}
impl Debug for AuthManager {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AuthManager")
.field("codex_home", &self.codex_home)
.field("inner", &self.inner)
.field("enable_codex_api_key_env", &self.enable_codex_api_key_env)
.field(
"auth_credentials_store_mode",
&self.auth_credentials_store_mode,
)
.field(
"forced_chatgpt_workspace_id",
&self.forced_chatgpt_workspace_id,
)
.field("has_external_auth", &self.has_external_auth())
.finish_non_exhaustive()
}
}
impl AuthManager {
@@ -1200,7 +1203,9 @@ impl AuthManager {
auth_credentials_store_mode: AuthCredentialsStoreMode::File,
forced_chatgpt_workspace_id: RwLock::new(None),
refresh_lock: AsyncMutex::new(()),
external_auth: RwLock::new(Some(ExternalAuth::Bearer(ExternalBearerAuth::new(config)))),
external_auth: RwLock::new(Some(
Arc::new(BearerTokenRefresher::new(config)) as Arc<dyn ExternalAuth>
)),
})
}
@@ -1223,7 +1228,7 @@ impl AuthManager {
/// For stale managed ChatGPT auth, first performs a guarded reload and then
/// refreshes only if the on-disk auth is unchanged.
pub async fn auth(&self) -> Option<CodexAuth> {
if let Some(auth) = self.resolve_external_bearer_auth().await {
if let Some(auth) = self.resolve_external_api_key_auth().await {
return Some(auth);
}
@@ -1346,13 +1351,13 @@ impl AuthManager {
}
}
pub fn set_external_chatgpt_auth_refresher(&self, refresher: Arc<dyn ExternalAuthRefresher>) {
pub fn set_external_auth(&self, external_auth: Arc<dyn ExternalAuth>) {
if let Ok(mut guard) = self.external_auth.write() {
*guard = Some(ExternalAuth::ChatgptRefresher(refresher));
*guard = Some(external_auth);
}
}
pub fn clear_external_chatgpt_auth_refresher(&self) {
pub fn clear_external_auth(&self) {
if let Ok(mut guard) = self.external_auth.write() {
*guard = None;
}
@@ -1371,12 +1376,8 @@ impl AuthManager {
.and_then(|guard| guard.clone())
}
pub fn has_external_chatgpt_auth_refresher(&self) -> bool {
self.external_auth
.read()
.ok()
.map(|guard| matches!(guard.as_ref(), Some(ExternalAuth::ChatgptRefresher(_))))
.unwrap_or(false)
pub fn has_external_auth(&self) -> bool {
self.external_auth().is_some()
}
pub fn is_external_chatgpt_auth_active(&self) -> bool {
@@ -1402,18 +1403,18 @@ impl AuthManager {
))
}
pub fn shared_with_external_chatgpt_auth_refresher(
pub fn shared_with_external_auth(
codex_home: PathBuf,
enable_codex_api_key_env: bool,
auth_credentials_store_mode: AuthCredentialsStoreMode,
refresher: Arc<dyn ExternalAuthRefresher>,
external_auth: Arc<dyn ExternalAuth>,
) -> Arc<Self> {
let manager = Self::shared(
codex_home,
enable_codex_api_key_env,
auth_credentials_store_mode,
);
manager.set_external_chatgpt_auth_refresher(refresher);
manager.set_external_auth(external_auth);
manager
}
@@ -1421,26 +1422,35 @@ impl AuthManager {
UnauthorizedRecovery::new(Arc::clone(self))
}
fn external_auth(&self) -> Option<ExternalAuth> {
fn external_auth(&self) -> Option<Arc<dyn ExternalAuth>> {
self.external_auth
.read()
.ok()
.and_then(|guard| guard.clone())
.and_then(|guard| guard.as_ref().cloned())
}
fn has_external_bearer_auth(&self) -> bool {
matches!(self.external_auth(), Some(ExternalAuth::Bearer(_)))
fn external_auth_mode(&self) -> Option<crate::AuthMode> {
self.external_auth()
.as_ref()
.map(|external_auth| external_auth.auth_mode())
}
async fn resolve_external_bearer_auth(&self) -> Option<CodexAuth> {
let ExternalAuth::Bearer(bearer_auth) = self.external_auth()? else {
fn has_external_api_key_auth(&self) -> bool {
self.external_auth_mode() == Some(crate::AuthMode::ApiKey)
}
async fn resolve_external_api_key_auth(&self) -> Option<CodexAuth> {
if !self.has_external_api_key_auth() {
return None;
};
}
match bearer_auth.resolve_access_token().await {
Ok(access_token) => Some(CodexAuth::from_api_key(&access_token)),
let external_auth = self.external_auth()?;
match external_auth.resolve().await {
Ok(Some(tokens)) => Some(CodexAuth::from_api_key(&tokens.access_token)),
Ok(None) => None,
Err(err) => {
tracing::error!("Failed to resolve external bearer auth: {err}");
tracing::error!("Failed to resolve external API key auth: {err}");
None
}
}
@@ -1534,14 +1544,14 @@ impl AuthManager {
}
pub fn get_api_auth_mode(&self) -> Option<ApiAuthMode> {
if self.has_external_bearer_auth() {
if self.has_external_api_key_auth() {
return Some(ApiAuthMode::ApiKey);
}
self.auth_cached().as_ref().map(CodexAuth::api_auth_mode)
}
pub fn auth_mode(&self) -> Option<crate::AuthMode> {
if self.has_external_bearer_auth() {
if self.has_external_api_key_auth() {
return Some(crate::AuthMode::ApiKey);
}
self.auth_cached().as_ref().map(CodexAuth::auth_mode)
@@ -1573,20 +1583,12 @@ impl AuthManager {
&self,
reason: ExternalAuthRefreshReason,
) -> Result<(), RefreshTokenError> {
if let Some(ExternalAuth::Bearer(bearer_auth)) = self.external_auth() {
return bearer_auth
.refresh_after_unauthorized()
.await
.map_err(RefreshTokenError::Transient);
}
let forced_chatgpt_workspace_id = self.forced_chatgpt_workspace_id();
let Some(ExternalAuth::ChatgptRefresher(refresher)) = self.external_auth() else {
let Some(external_auth) = self.external_auth() else {
return Err(RefreshTokenError::Transient(std::io::Error::other(
"external auth refresher is not configured",
"external auth is not configured",
)));
};
let forced_chatgpt_workspace_id = self.forced_chatgpt_workspace_id();
let previous_account_id = self
.auth_cached()
.as_ref()
@@ -1596,7 +1598,13 @@ impl AuthManager {
previous_account_id,
};
let refreshed = refresher.refresh(context).await?;
let refreshed = external_auth
.refresh(context)
.await
.map_err(RefreshTokenError::Transient)?;
if external_auth.auth_mode() == crate::AuthMode::ApiKey {
return Ok(());
}
let Some(chatgpt_metadata) = refreshed.chatgpt_metadata() else {
return Err(RefreshTokenError::Transient(std::io::Error::other(
"external auth refresh did not return ChatGPT metadata",
codex-rs/login/src/lib.rs
+1 -1
View File
@@ -22,10 +22,10 @@ pub use auth::AuthManager;
pub use auth::CLIENT_ID;
pub use auth::CODEX_API_KEY_ENV_VAR;
pub use auth::CodexAuth;
pub use auth::ExternalAuth;
pub use auth::ExternalAuthChatgptMetadata;
pub use auth::ExternalAuthRefreshContext;
pub use auth::ExternalAuthRefreshReason;
pub use auth::ExternalAuthRefresher;
pub use auth::ExternalAuthTokens;
pub use auth::OPENAI_API_KEY_ENV_VAR;
pub use auth::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
codex-rs/mcp-server/tests/common/Cargo.toml
+3
View File
@@ -7,6 +7,9 @@ license.workspace = true
[lib]
path = "lib.rs"
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
codex-core = { workspace = true }
codex-rs/network-proxy/Cargo.toml
+2 -2
View File
@@ -1,7 +1,7 @@
[package]
name = "codex-network-proxy"
edition = "2024"
version = { workspace = true }
edition.workspace = true
version.workspace = true
license.workspace = true
[lib]
codex-rs/shell-escalation/Cargo.toml
+3
View File
@@ -8,6 +8,9 @@ version.workspace = true
name = "codex-execve-wrapper"
path = "src/bin/main_execve_wrapper.rs"
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
async-trait = { workspace = true }
codex-rs/shell-escalation/src/unix/escalate_server.rs
+1 -1
View File
@@ -601,7 +601,7 @@ mod tests {
let execve_wrapper_str = execve_wrapper.to_string_lossy().to_string();
let server = EscalateServer::new(
PathBuf::from("/bin/zsh"),
execve_wrapper.clone(),
execve_wrapper,
DeterministicEscalationPolicy {
decision: EscalationDecision::run(),
},
codex-rs/shell-escalation/src/unix/socket.rs
+1 -1
View File
@@ -62,7 +62,7 @@ fn extract_fds(control: &[u8]) -> Vec<OwnedFd> {
let fd_count: usize = {
// `cmsghdr::cmsg_len` is not typed consistently across targets, so normalize it
// before doing the size arithmetic.
#[allow(clippy::useless_conversion)]
#[allow(clippy::useless_conversion, clippy::expect_used)]
let cmsg_data_len = usize::try_from(unsafe { (*cmsg).cmsg_len })
.expect("cmsghdr length fits")
- unsafe { libc::CMSG_LEN(0) as usize };
codex-rs/utils/pty/Cargo.toml
+1 -1
View File
@@ -1,5 +1,5 @@
[package]
edition = "2024"
edition.workspace = true
license.workspace = true
name = "codex-utils-pty"
version.workspace = true
codex-rs/windows-sandbox-rs/Cargo.toml
+4 -1
View File
@@ -1,6 +1,6 @@
[package]
build = "build.rs"
edition = "2024"
edition.workspace = true
license.workspace = true
name = "codex-windows-sandbox"
version.workspace = true
@@ -17,6 +17,9 @@ path = "src/bin/setup_main.rs"
name = "codex-command-runner"
path = "src/bin/command_runner.rs"
[lints]
workspace = true
[dependencies]
anyhow = "1.0"
base64 = { workspace = true }