diff --git a/.github/scripts/select_windows_bazel_targets.py b/.github/scripts/select_windows_bazel_targets.py new file mode 100644 index 000000000000..226bd7040298 --- /dev/null +++ b/.github/scripts/select_windows_bazel_targets.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 + +"""Assign Windows Bazel test targets using checked-in duration estimates.""" + +import argparse +import sys +from pathlib import Path + + +DEFAULT_DURATION_SECONDS = 1 +DURATIONS_PATH = Path(__file__).with_name("windows_bazel_test_durations.tsv") + + +def read_durations(path: Path) -> dict[str, int]: + durations = {} + for line_number, line in enumerate( + path.read_text(encoding="utf-8").splitlines(), 1 + ): + if not line.strip() or line.lstrip().startswith("#"): + continue + fields = line.split("\t") + if len(fields) != 2 or not fields[0] or not fields[1].isdecimal(): + raise ValueError( + f"{path}:{line_number}: expected targetpositive integer" + ) + target, duration_text = fields + duration = int(duration_text) + if duration < 1: + raise ValueError(f"{path}:{line_number}: duration must be positive") + if target in durations: + raise ValueError(f"{path}:{line_number}: duplicate target {target}") + durations[target] = duration + return durations + + +def assign_targets( + targets: list[str], durations: dict[str, int], shard_count: int +) -> tuple[list[list[str]], list[int]]: + if shard_count < 1: + raise ValueError("shard count must be positive") + if not targets: + raise ValueError("no Bazel test targets were provided") + if any(not target.strip() for target in targets): + raise ValueError("Bazel test target labels must not be empty") + if len(targets) != len(set(targets)): + raise ValueError("duplicate Bazel test target labels were provided") + + assignments: list[list[str]] = [[] for _ in range(shard_count)] + loads = [0] * shard_count + # Place the longest targets first, breaking ties by label and shard number. + for target in sorted( + targets, + key=lambda target: (-durations.get(target, DEFAULT_DURATION_SECONDS), target), + ): + shard = min(range(shard_count), key=lambda shard: (loads[shard], shard)) + assignments[shard].append(target) + loads[shard] += durations.get(target, DEFAULT_DURATION_SECONDS) + for assignment in assignments: + assignment.sort() + return assignments, loads + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--shard", required=True, type=int) + parser.add_argument("--shard-count", required=True, type=int) + parser.add_argument( + "--durations", + type=Path, + default=DURATIONS_PATH, + help="use this timing file instead of the checked-in default", + ) + args = parser.parse_args() + if not 1 <= args.shard <= args.shard_count: + parser.error("shard must be between 1 and shard count") + + try: + durations = read_durations(args.durations) + targets = sys.stdin.read().splitlines() + assignments, loads = assign_targets(targets, durations, args.shard_count) + except (OSError, ValueError) as error: + parser.error(str(error)) + + selected_targets = assignments[args.shard - 1] + if not selected_targets: + parser.error( + f"no Bazel test targets selected for shard {args.shard}/{args.shard_count}" + ) + + target_set = set(targets) + print( + f"Windows Bazel shards: targets={[len(assignment) for assignment in assignments]}, " + f"estimated test-seconds={loads}, default weights={len(target_set - durations.keys())}, " + f"unused weights={len(durations.keys() - target_set)}. " + f"Selected shard {args.shard}/{args.shard_count}.", + file=sys.stderr, + ) + # Keep LF delimiters when native Windows Python feeds Git Bash mapfile. + sys.stdout.buffer.write(("\n".join(selected_targets) + "\n").encode("utf-8")) + + +if __name__ == "__main__": + main() diff --git a/.github/scripts/test_select_windows_bazel_targets.py b/.github/scripts/test_select_windows_bazel_targets.py new file mode 100644 index 000000000000..0c708ba4c5a2 --- /dev/null +++ b/.github/scripts/test_select_windows_bazel_targets.py @@ -0,0 +1,147 @@ +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from select_windows_bazel_targets import DURATIONS_PATH +from select_windows_bazel_targets import assign_targets +from select_windows_bazel_targets import read_durations + + +class SelectWindowsBazelTargetsTest(unittest.TestCase): + def test_balances_targets_with_complete_deterministic_coverage(self) -> None: + durations = { + f"//test:{name}": duration + for name, duration in zip("abcdefgh", range(8, 0, -1)) + } + targets = list(durations) + + assignments, loads = assign_targets(targets, durations, 4) + + self.assertEqual(loads, [9, 9, 9, 9]) + self.assertEqual( + sorted(target for shard in assignments for target in shard), sorted(targets) + ) + self.assertEqual( + assign_targets(list(reversed(targets)), durations, 4), (assignments, loads) + ) + self.assertTrue(all(shard == sorted(shard) for shard in assignments)) + + def test_missing_weights_use_one_and_unused_weights_do_not_create_targets( + self, + ) -> None: + targets = ["//test:c", "//test:a", "//test:b"] + + assignments, loads = assign_targets(targets, {"//test:stale": 100}, 2) + + self.assertEqual(assignments, [["//test:a", "//test:c"], ["//test:b"]]) + self.assertEqual(loads, [2, 1]) + + def test_rejects_empty_or_duplicate_target_input(self) -> None: + for targets in ([], [""], ["//test:a", "//test:a"]): + with self.subTest(targets=targets), self.assertRaises(ValueError): + assign_targets(targets, {}, 4) + + def test_reads_duration_file_and_rejects_invalid_weights(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "durations.tsv" + path.write_text( + "# comment\n\n//test:a\t12\n//test:b\t1\n", encoding="utf-8" + ) + self.assertEqual(read_durations(path), {"//test:a": 12, "//test:b": 1}) + + for contents in ( + "//test:a 12\n", + "//test:a\t0\n", + "//test:a\t-1\n", + "//test:a\t1.5\n", + "//test:a\t1\n//test:a\t2\n", + ): + with self.subTest(contents=contents), self.assertRaises(ValueError): + path.write_text(contents, encoding="utf-8") + read_durations(path) + + def test_cli_uses_checked_in_weights_and_writes_only_lf_delimited_targets( + self, + ) -> None: + durations = read_durations(DURATIONS_PATH) + known = sorted(durations, key=lambda target: (-durations[target], target))[:4] + self.assertEqual(len(known), 4) + targets = [*known, "//test:new-target"] + expected, _ = assign_targets(targets, durations, 4) + script = DURATIONS_PATH.with_name("select_windows_bazel_targets.py") + + for shard, assignment in enumerate(expected, 1): + command = [ + sys.executable, + str(script), + "--shard", + str(shard), + "--shard-count", + "4", + ] + for ordered_targets in (targets, list(reversed(targets))): + with self.subTest(shard=shard, ordered_targets=ordered_targets): + result = subprocess.run( + command, + input=("\r\n".join(ordered_targets) + "\r\n").encode(), + capture_output=True, + check=True, + ) + self.assertEqual( + result.stdout, ("\n".join(assignment) + "\n").encode() + ) + self.assertIn(b"default weights=1", result.stderr) + + def test_cli_uses_only_the_requested_duration_file(self) -> None: + script = DURATIONS_PATH.with_name("select_windows_bazel_targets.py").resolve() + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "durations.tsv" + path.write_text("//test:a\t1\n//test:b\t9\n", encoding="utf-8") + command = [sys.executable, str(script), "--shard-count", "2"] + for shard, expected in ((1, b"//test:b\n"), (2, b"//test:a\n//test:c\n")): + with self.subTest(shard=shard): + result = subprocess.run( + [*command, "--shard", str(shard), "--durations", path.name], + input=b"//test:a\n//test:b\n//test:c\n", + cwd=temp_dir, + capture_output=True, + check=True, + ) + self.assertEqual(result.stdout, expected) + self.assertIn(b"estimated test-seconds=[9, 2]", result.stderr) + self.assertIn(b"unused weights=0", result.stderr) + + result = subprocess.run( + [*command, "--shard", "1", "--durations", "missing.tsv"], + input=b"//test:a\n", + cwd=temp_dir, + capture_output=True, + check=False, + ) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(result.stdout, b"") + self.assertIn(b"missing.tsv", result.stderr) + + def test_cli_rejects_bad_or_empty_selection_without_partial_output(self) -> None: + script = DURATIONS_PATH.with_name("select_windows_bazel_targets.py") + for targets, error in ( + (b"", b"no Bazel test targets were provided"), + (b"//test:a\n//test:a\n", b"duplicate Bazel test target labels"), + (b"//test:a\n", b"no Bazel test targets selected for shard 2/2"), + ): + with self.subTest(targets=targets): + result = subprocess.run( + [sys.executable, str(script), "--shard", "2", "--shard-count", "2"], + input=targets, + capture_output=True, + check=False, + ) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(result.stdout, b"") + self.assertIn(error, result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/scripts/windows_bazel_test_durations.tsv b/.github/scripts/windows_bazel_test_durations.tsv new file mode 100644 index 000000000000..32bcc2478347 --- /dev/null +++ b/.github/scripts/windows_bazel_test_durations.tsv @@ -0,0 +1,253 @@ +# Windows Bazel duration weights; columns: targetseconds. +# Seeded from 24 successful public openai/codex Windows job logs across six main +# runs, September 21-23, 2026. Public GitHub Actions run IDs: +# 35638529858, 35665540272, 35694732932, 35912411800, 35919297566, 35924169500. +# To refresh, take the median summed inner-run time (N * average) from clean, +# uncached PASSED summaries, excluding partial cache hits and FLAKY results. +# Round up to positive integer seconds. These weights exclude build/download time. +# Every queried target is assigned; missing weights default to 1 and stale entries are ignored. +//codex-rs/agent-graph-store:agent-graph-store-unit-tests 76 +//codex-rs/agent-identity:agent-identity-unit-tests 66 +//codex-rs/agent-roles:agent-roles-unit-tests 48 +//codex-rs/analytics:analytics-unit-tests 29 +//codex-rs/app-server-client:app-server-client-unit-tests 51 +//codex-rs/app-server-daemon:app-server-daemon-unit-tests 4 +//codex-rs/app-server-protocol:app-server-protocol-unit-tests 19 +//codex-rs/app-server-test-client:app-server-test-client-unit-tests 25 +//codex-rs/app-server-test-client:codex-app-server-test-client-bin-unit-tests 1 +//codex-rs/app-server-transport:app-server-transport-unit-tests 65 +//codex-rs/app-server/tests/common:common-unit-tests 26 +//codex-rs/app-server:app-server-all-test-windows-cross 2835 +//codex-rs/app-server:app-server-unit-tests 399 +//codex-rs/app-server:codex-app-server-bin-unit-tests 48 +//codex-rs/app-server:codex-app-server-test-notify-capture-bin-unit-tests 1 +//codex-rs/app-server:exec-server-bin-unit-tests 1 +//codex-rs/app-server:test_notify_capture-bin-unit-tests 1 +//codex-rs/apply-patch:apply-patch-all-test-windows-cross 50 +//codex-rs/apply-patch:apply-patch-unit-tests 48 +//codex-rs/apply-patch:apply_patch-bin-unit-tests 24 +//codex-rs/arg0:arg0-unit-tests 24 +//codex-rs/backend-client:backend-client-unit-tests 41 +//codex-rs/chatgpt:chatgpt-all-test-windows-cross 26 +//codex-rs/chatgpt:chatgpt-unit-tests 27 +//codex-rs/cli:cli-app_server-test-windows-cross 3 +//codex-rs/cli:cli-app_server_daemon-test-windows-cross 1 +//codex-rs/cli:cli-cloud_auth-test-windows-cross 47 +//codex-rs/cli:cli-cloud_config-test-windows-cross 3 +//codex-rs/cli:cli-daemon_startup-test-windows-cross 4 +//codex-rs/cli:cli-daemon_telemetry-test-windows-cross 1 +//codex-rs/cli:cli-debug_clear_memories-test-windows-cross 4 +//codex-rs/cli:cli-debug_models-test-windows-cross 2 +//codex-rs/cli:cli-delete-test-windows-cross 2 +//codex-rs/cli:cli-doctor_enterprise_network-test-windows-cross 3 +//codex-rs/cli:cli-doctor_path_safety-test-windows-cross 47 +//codex-rs/cli:cli-exec_server-test-windows-cross 39 +//codex-rs/cli:cli-exec_server_websocket_auth-test-windows-cross 4 +//codex-rs/cli:cli-execpolicy-test-windows-cross 2 +//codex-rs/cli:cli-features-test-windows-cross 3 +//codex-rs/cli:cli-login-test-windows-cross 4 +//codex-rs/cli:cli-marketplace_add-test-windows-cross 2 +//codex-rs/cli:cli-marketplace_remove-test-windows-cross 1 +//codex-rs/cli:cli-marketplace_upgrade-test-windows-cross 2 +//codex-rs/cli:cli-mcp_add_remove-test-windows-cross 3 +//codex-rs/cli:cli-mcp_list-test-windows-cross 2 +//codex-rs/cli:cli-mcp_login-test-windows-cross 2 +//codex-rs/cli:cli-plugin_cli-test-windows-cross 7 +//codex-rs/cli:cli-queue-test-windows-cross 1 +//codex-rs/cli:cli-sandbox_cloud_config-test-windows-cross 48 +//codex-rs/cli:cli-sandbox_network_proxy-test-windows-cross 1 +//codex-rs/cli:cli-sandbox_tty-test-windows-cross 1 +//codex-rs/cli:cli-unit-tests 2 +//codex-rs/cli:cli-update-test-windows-cross 47 +//codex-rs/cli:cli-worktree-test-windows-cross 12 +//codex-rs/cli:codex-bin-unit-tests 60 +//codex-rs/cli:logs_client-bin-unit-tests 1 +//codex-rs/cloud-config:cloud-config-unit-tests 47 +//codex-rs/cloud-tasks-client:cloud-tasks-client-unit-tests 35 +//codex-rs/cloud-tasks-mock-client:cloud-tasks-mock-client-unit-tests 59 +//codex-rs/cloud-tasks:cloud-tasks-env_filter-test-windows-cross 1 +//codex-rs/cloud-tasks:cloud-tasks-unit-tests 1 +//codex-rs/code-mode-host:code-mode-host-grpc-test-windows-cross 28 +//codex-rs/code-mode-host:code-mode-host-grpc_notifications-test-windows-cross 25 +//codex-rs/code-mode-host:code-mode-host-grpc_tcp-test-windows-cross 1 +//codex-rs/code-mode-host:code-mode-host-grpc_timing-test-windows-cross 7 +//codex-rs/code-mode-host:code-mode-host-stdio-test-windows-cross 11 +//codex-rs/code-mode-host:code-mode-host-unit-tests 2 +//codex-rs/code-mode-host:codex-code-mode-host-bin-unit-tests 1 +//codex-rs/code-mode-protocol:code-mode-protocol-unit-tests 62 +//codex-rs/code-mode-runtime:code-mode-runtime-array_sort-test-windows-cross 24 +//codex-rs/code-mode-runtime:code-mode-runtime-jit-test-windows-cross 1 +//codex-rs/code-mode-runtime:code-mode-runtime-unit-tests 7 +//codex-rs/code-mode:code-mode-unit-tests 49 +//codex-rs/codex-api:codex-api-clients-test-windows-cross 75 +//codex-rs/codex-api:codex-api-models_integration-test-windows-cross 66 +//codex-rs/codex-api:codex-api-realtime_websocket_e2e-test-windows-cross 63 +//codex-rs/codex-api:codex-api-realtime_websocket_tls-test-windows-cross 66 +//codex-rs/codex-api:codex-api-sse_end_to_end-test-windows-cross 48 +//codex-rs/codex-api:codex-api-unit-tests 60 +//codex-rs/codex-home:codex-home-unit-tests 13 +//codex-rs/codex-mcp:codex-mcp-unit-tests 61 +//codex-rs/config-schema:codex-write-config-schema-bin-unit-tests 48 +//codex-rs/config:config-unit-tests 75 +//codex-rs/connectors:connectors-unit-tests 39 +//codex-rs/context-fragments:context-fragments-unit-tests 59 +//codex-rs/core-api:core-api-unit-tests 1 +//codex-rs/core-plugins:core-plugins-unit-tests 20 +//codex-rs/core/tests/common:common-unit-tests 48 +//codex-rs/core:core-all-test-windows-cross 3162 +//codex-rs/core:core-mcp_trace_propagation-test-windows-cross 2 +//codex-rs/core:core-responses_headers-test-windows-cross 5 +//codex-rs/core:core-unit-tests 251 +//codex-rs/exec-server-protocol:exec-server-protocol-unit-tests 48 +//codex-rs/exec-server/tests/support:support-unit-tests 48 +//codex-rs/exec-server:exec-server-accepted_websocket-test-windows-cross 1 +//codex-rs/exec-server:exec-server-capability_discovery-test-windows-cross 1 +//codex-rs/exec-server:exec-server-chatgpt_cloudflare_affinity-test-windows-cross 1 +//codex-rs/exec-server:exec-server-deferred_environment-test-windows-cross 1 +//codex-rs/exec-server:exec-server-environment-test-windows-cross 10 +//codex-rs/exec-server:exec-server-environment_config-test-windows-cross 1 +//codex-rs/exec-server:exec-server-exec_process-test-windows-cross 1 +//codex-rs/exec-server:exec-server-file_stream-test-windows-cross 1 +//codex-rs/exec-server:exec-server-file_system_unix-test-windows-cross 1 +//codex-rs/exec-server:exec-server-file_system_windows-test-windows-cross 11 +//codex-rs/exec-server:exec-server-forward-test-windows-cross 10 +//codex-rs/exec-server:exec-server-health-test-windows-cross 1 +//codex-rs/exec-server:exec-server-http_client-test-windows-cross 2 +//codex-rs/exec-server:exec-server-http_request-test-windows-cross 1 +//codex-rs/exec-server:exec-server-http_request_logging-test-windows-cross 3 +//codex-rs/exec-server:exec-server-initialize-test-windows-cross 1 +//codex-rs/exec-server:exec-server-macos_fs_spawn-test-windows-cross 1 +//codex-rs/exec-server:exec-server-process-test-windows-cross 4 +//codex-rs/exec-server:exec-server-relay-test-windows-cross 8 +//codex-rs/exec-server:exec-server-selected_capability_roots-test-windows-cross 1 +//codex-rs/exec-server:exec-server-unit-tests 116 +//codex-rs/exec-server:exec-server-websocket-test-windows-cross 1 +//codex-rs/exec:codex-exec-bin-unit-tests 1 +//codex-rs/exec:exec-all-test-windows-cross 331 +//codex-rs/exec:exec-event_processor_with_json_output-test-windows-cross 1 +//codex-rs/exec:exec-unit-tests 47 +//codex-rs/ext/agent-message-board:agent-message-board-local_board-test-windows-cross 46 +//codex-rs/ext/agent-message-board:agent-message-board-unit-tests 30 +//codex-rs/ext/agent:agent-agent_service-test-windows-cross 3 +//codex-rs/ext/agent:agent-unit-tests 1 +//codex-rs/ext/connectors:connectors-unit-tests 47 +//codex-rs/ext/extension-api:extension-api-capabilities-test-windows-cross 14 +//codex-rs/ext/extension-api:extension-api-registry-test-windows-cross 31 +//codex-rs/ext/extension-api:extension-api-state-test-windows-cross 57 +//codex-rs/ext/extension-api:extension-api-unit-tests 35 +//codex-rs/ext/git-attribution:git-attribution-unit-tests 30 +//codex-rs/ext/goal:goal-accounting-test-windows-cross 1 +//codex-rs/ext/goal:goal-goal_extension_backend-test-windows-cross 9 +//codex-rs/ext/goal:goal-steering-test-windows-cross 1 +//codex-rs/ext/goal:goal-unit-tests 47 +//codex-rs/ext/guardian-reviewer:guardian-reviewer-unit-tests 36 +//codex-rs/ext/guardian-v2:guardian-v2-unit-tests 50 +//codex-rs/ext/history-notes:history-notes-history_notes_extension-test-windows-cross 1 +//codex-rs/ext/history-notes:history-notes-image_model_input-test-windows-cross 2 +//codex-rs/ext/history-notes:history-notes-unit-tests 1 +//codex-rs/ext/image-generation:image-generation-unit-tests 47 +//codex-rs/ext/mcp:mcp-hosted_apps_mcp-test-windows-cross 26 +//codex-rs/ext/mcp:mcp-plugin_mcp-test-windows-cross 5 +//codex-rs/ext/mcp:mcp-unit-tests 48 +//codex-rs/ext/memories:memories-unit-tests 28 +//codex-rs/ext/queue:queue-queue_service-test-windows-cross 9 +//codex-rs/ext/queue:queue-unit-tests 25 +//codex-rs/ext/skills:skills-executor_file_system_authority-test-windows-cross 36 +//codex-rs/ext/skills:skills-skills_extension-test-windows-cross 7 +//codex-rs/ext/skills:skills-unit-tests 8 +//codex-rs/ext/web-search:web-search-unit-tests 1 +//codex-rs/external-agent-migration:external-agent-migration-unit-tests 4 +//codex-rs/features:features-unit-tests 47 +//codex-rs/feedback:feedback-unit-tests 77 +//codex-rs/file-system:file-system-unit-tests 48 +//codex-rs/git-utils:git-utils-unit-tests 83 +//codex-rs/guardian-context:guardian-context-cache_prefix-test-windows-cross 66 +//codex-rs/guardian-context:guardian-context-unit-tests 67 +//codex-rs/history:history-unit-tests 59 +//codex-rs/hooks:hooks-unit-tests 16 +//codex-rs/hooks:write_hooks_schema_fixtures-bin-unit-tests 48 +//codex-rs/lmstudio:lmstudio-unit-tests 1 +//codex-rs/login:login-all-test-windows-cross 5 +//codex-rs/login:login-unit-tests 320 +//codex-rs/memories/read:read-unit-tests 62 +//codex-rs/memories/write:write-unit-tests 20 +//codex-rs/message-history:message-history-unit-tests 74 +//codex-rs/model-provider-info:model-provider-info-unit-tests 63 +//codex-rs/model-provider:model-provider-unit-tests 64 +//codex-rs/models-manager:models-manager-unit-tests 35 +//codex-rs/mxc-sandbox:mxc-sandbox-unit-tests 36 +//codex-rs/network-proxy:codex-network-proxy-bin-unit-tests 48 +//codex-rs/network-proxy:network-proxy-unit-tests 85 +//codex-rs/network-proxy:network-proxy-windows_stable_ingress-test-windows-cross 37 +//codex-rs/ollama:ollama-unit-tests 1 +//codex-rs/otel:otel-tests-test-windows-cross 17 +//codex-rs/otel:otel-unit-tests 48 +//codex-rs/plugin:plugin-unit-tests 74 +//codex-rs/prompts:prompts-unit-tests 62 +//codex-rs/protocol:protocol-unit-tests 67 +//codex-rs/response-debug-context:response-debug-context-unit-tests 66 +//codex-rs/rmcp-client:rmcp-client-foreign_stdio_cwd-test-windows-cross 48 +//codex-rs/rmcp-client:rmcp-client-mcp_2026_discovery-test-windows-cross 4 +//codex-rs/rmcp-client:rmcp-client-mcp_2026_message_limits-test-windows-cross 4 +//codex-rs/rmcp-client:rmcp-client-mcp_2026_mrtr-test-windows-cross 2 +//codex-rs/rmcp-client:rmcp-client-mcp_2026_oauth_discovery-test-windows-cross 1 +//codex-rs/rmcp-client:rmcp-client-mcp_2026_sse_discovery-test-windows-cross 1 +//codex-rs/rmcp-client:rmcp-client-mcp_2026_stdio-test-windows-cross 39 +//codex-rs/rmcp-client:rmcp-client-mcp_2026_stdio_discovery-test-windows-cross 2 +//codex-rs/rmcp-client:rmcp-client-mcp_events-test-windows-cross 2 +//codex-rs/rmcp-client:rmcp-client-process_group_cleanup-test-windows-cross 1 +//codex-rs/rmcp-client:rmcp-client-resources-test-windows-cross 1 +//codex-rs/rmcp-client:rmcp-client-stdio_descriptor_boundary-test-windows-cross 1 +//codex-rs/rmcp-client:rmcp-client-stdio_message_limits-test-windows-cross 3 +//codex-rs/rmcp-client:rmcp-client-stdio_stderr_cleanup-test-windows-cross 1 +//codex-rs/rmcp-client:rmcp-client-streamable_http_oauth_startup-test-windows-cross 6 +//codex-rs/rmcp-client:rmcp-client-streamable_http_oauth_store_pinning-test-windows-cross 38 +//codex-rs/rmcp-client:rmcp-client-streamable_http_oauth_tool_call-test-windows-cross 2 +//codex-rs/rmcp-client:rmcp-client-streamable_http_recovery-test-windows-cross 51 +//codex-rs/rmcp-client:rmcp-client-streamable_http_remote-test-windows-cross 3 +//codex-rs/rmcp-client:rmcp-client-streamable_http_test_support-test-windows-cross 1 +//codex-rs/rmcp-client:rmcp-client-streamable_http_user_agent-test-windows-cross 1 +//codex-rs/rmcp-client:rmcp-client-unit-tests 50 +//codex-rs/rmcp-client:rmcp_test_server-bin-unit-tests 39 +//codex-rs/rmcp-client:test_mcp_2026_discovery_stdio_server-bin-unit-tests 43 +//codex-rs/rmcp-client:test_mcp_2026_stdio_server-bin-unit-tests 25 +//codex-rs/rmcp-client:test_stdio_server-bin-unit-tests 41 +//codex-rs/rmcp-client:test_streamable_http_server-bin-unit-tests 40 +//codex-rs/rollout-trace:rollout-trace-unit-tests 76 +//codex-rs/rollout:rollout-unit-tests 56 +//codex-rs/sandboxing:sandboxing-unit-tests 74 +//codex-rs/secrets:secrets-unit-tests 57 +//codex-rs/shell-command:shell-command-unit-tests 60 +//codex-rs/shell-escalation:codex-execve-wrapper-bin-unit-tests 62 +//codex-rs/shell-escalation:shell-escalation-unit-tests 60 +//codex-rs/skills:skills-unit-tests 59 +//codex-rs/state:state-unit-tests 242 +//codex-rs/test-binary-support:test-binary-support-unit-tests 48 +//codex-rs/thread-manager-sample:codex-thread-manager-sample-bin-unit-tests 47 +//codex-rs/thread-store:thread-store-unit-tests 95 +//codex-rs/tools:tools-json_schema_policy_fixtures-test-windows-cross 34 +//codex-rs/tools:tools-unit-tests 37 +//codex-rs/tui:codex-tui-bin-unit-tests 2 +//codex-rs/tui:md-events-bin-unit-tests 1 +//codex-rs/tui:tui-all-test-windows-cross 36 +//codex-rs/tui:tui-manager_dependency_regression-test-windows-cross 2 +//codex-rs/tui:tui-test_backend-test-windows-cross 25 +//codex-rs/tui:tui-unit-tests 1003 +//codex-rs/utils/approval-presets:approval-presets-unit-tests 66 +//codex-rs/utils/audio:audio-unit-tests 76 +//codex-rs/utils/cli:cli-unit-tests 64 +//codex-rs/utils/git-discovery:git-discovery-unit-tests 59 +//codex-rs/utils/oss:oss-unit-tests 1 +//codex-rs/utils/output-truncation:output-truncation-unit-tests 60 +//codex-rs/utils/plugins:plugins-unit-tests 36 +//codex-rs/utils/sandbox-summary:sandbox-summary-unit-tests 66 +//codex-rs/windows-sandbox-rs:codex-command-runner-bin-unit-tests 1 +//codex-rs/windows-sandbox-rs:codex-windows-managed-deny-probe-bin-unit-tests 1 +//codex-rs/windows-sandbox-rs:codex-windows-sandbox-setup-bin-unit-tests 1 +//codex-rs/windows-sandbox-rs:windows-sandbox-rs-helper_manifest-test-windows-cross 1 +//codex-rs/windows-sandbox-rs:windows-sandbox-rs-unit-tests 23 +//codex-rs/windows-sandbox-service:codex-windows-sandbox-service-bin-unit-tests 46 +//codex-rs/windows-sandbox-service:windows-sandbox-service-unit-tests 12 +//codex-rs/worktree:worktree-settings-test-windows-cross 59 +//codex-rs/worktree:worktree-unit-tests 76 +//codex-rs/worktree:worktree-worktree-test-windows-cross 92 diff --git a/.github/workflows/bazel.yml b/.github/workflows/bazel.yml index 568086a93d7a..ede5a7f07032 100644 --- a/.github/workflows/bazel.yml +++ b/.github/workflows/bazel.yml @@ -160,6 +160,11 @@ jobs: shell: pwsh run: python .github/scripts/test_run_bazel_with_buildbuddy.py + - name: Test Windows Bazel target selector + if: matrix.shard == 1 + shell: pwsh + run: python -m unittest discover -s .github/scripts -p test_select_windows_bazel_targets.py + - name: Prepare Bazel CI id: prepare_bazel uses: ./.github/actions/prepare-bazel-ci @@ -180,29 +185,19 @@ jobs: set -euo pipefail bazel_test_query='tests(//...) except tests(//third_party/v8:all) except attr(tags, "manual", tests(//...))' - mapfile -t bazel_targets < <( - ./.github/scripts/run-bazel-query-ci.sh --output=label -- "${bazel_test_query}" \ - | LC_ALL=C sort - ) - - selected_targets=() - for bazel_target in "${bazel_targets[@]}"; do - target_bucket="$( - printf '%s\n' "${bazel_target}" \ - | cksum \ - | awk -v shard_count="${BAZEL_TEST_SHARD_COUNT}" '{ print ($1 % shard_count) + 1 }' - )" - if [[ "${target_bucket}" == "${BAZEL_TEST_SHARD}" ]]; then - selected_targets+=("${bazel_target}") - fi - done - - if [[ ${#selected_targets[@]} -eq 0 ]]; then - echo "No Bazel test targets selected for Windows shard ${BAZEL_TEST_SHARD}/${BAZEL_TEST_SHARD_COUNT}." >&2 - exit 1 - fi - - echo "Selected ${#selected_targets[@]} of ${#bazel_targets[@]} Bazel test targets for Windows shard ${BAZEL_TEST_SHARD}/${BAZEL_TEST_SHARD_COUNT}." + bazel_targets_file="$(mktemp)" + selected_targets_file="$(mktemp)" + trap 'rm -f "${bazel_targets_file}" "${selected_targets_file}"' EXIT + + # Keep query and selection in the foreground so failures cannot leave + # different jobs assigning different partial target lists. + ./.github/scripts/run-bazel-query-ci.sh --output=label -- "${bazel_test_query}" \ + | LC_ALL=C sort > "${bazel_targets_file}" + python .github/scripts/select_windows_bazel_targets.py \ + --shard "${BAZEL_TEST_SHARD}" \ + --shard-count "${BAZEL_TEST_SHARD_COUNT}" \ + < "${bazel_targets_file}" > "${selected_targets_file}" + mapfile -t selected_targets < "${selected_targets_file}" bazel_test_args=( test diff --git a/MODULE.bazel b/MODULE.bazel index 001bb19ff7ed..e1a6a7ff0563 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -191,6 +191,7 @@ single_version_override( "//patches:rules_rs_build_script_deps_annotation.patch", "//patches:rules_rs_rust_debug_modes.patch", "//patches:rules_rs_windows_msvc_linker.patch", + "//patches:rules_rs_workspace_package_version.patch", "//patches:rules_rs_zlib_snapshot_urls.patch", ], version = "0.0.96", diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock index 8e1990f965e1..ea12d29eead6 100644 --- a/MODULE.bazel.lock +++ b/MODULE.bazel.lock @@ -1161,7 +1161,7 @@ "gstreamer_0.25.3": "{\"dependencies\":[{\"name\":\"cfg-if\",\"req\":\"^1.0\"},{\"name\":\"futures-channel\",\"req\":\"^0.3\"},{\"name\":\"futures-core\",\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"futures-executor\",\"req\":\"^0.3.1\"},{\"default_features\":false,\"name\":\"futures-util\",\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"gir-format-check\",\"req\":\"^0.1\"},{\"name\":\"glib\",\"req\":\"^0.22.0\"},{\"name\":\"gstreamer-sys\",\"req\":\"^0.25.0\"},{\"name\":\"itertools\",\"req\":\"^0.15\"},{\"name\":\"kstring\",\"req\":\"^2.0\"},{\"name\":\"libc\",\"req\":\"^0.2\"},{\"name\":\"log\",\"optional\":true,\"req\":\"^0.4\"},{\"name\":\"muldiv\",\"req\":\"^1\"},{\"default_features\":false,\"name\":\"num-integer\",\"req\":\"^0.1\"},{\"default_features\":false,\"name\":\"num-rational\",\"req\":\"^0.4\"},{\"name\":\"opt-ops\",\"package\":\"option-operations\",\"req\":\"^0.6\"},{\"name\":\"pastey\",\"req\":\"^0.2\"},{\"name\":\"pin-project-lite\",\"req\":\"^0.2\"},{\"kind\":\"dev\",\"name\":\"ron\",\"req\":\"^0.12\"},{\"features\":[\"derive\"],\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0\"},{\"name\":\"serde_bytes\",\"optional\":true,\"req\":\"^0.11\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0\"},{\"features\":[\"write\"],\"name\":\"smallvec\",\"req\":\"^1.0\"},{\"name\":\"thiserror\",\"req\":\"^2\"}],\"features\":{\"default\":[],\"log\":[\"dep:log\"],\"serde\":[\"num-rational/serde\",\"dep:serde\",\"serde_bytes\"],\"v1_16\":[\"gstreamer-sys/v1_16\"],\"v1_18\":[\"gstreamer-sys/v1_18\",\"v1_16\"],\"v1_20\":[\"gstreamer-sys/v1_20\",\"v1_18\"],\"v1_22\":[\"gstreamer-sys/v1_22\",\"v1_20\"],\"v1_24\":[\"gstreamer-sys/v1_24\",\"v1_22\"],\"v1_26\":[\"gstreamer-sys/v1_26\",\"v1_24\"],\"v1_28\":[\"gstreamer-sys/v1_28\",\"v1_26\"],\"v1_30\":[\"gstreamer-sys/v1_30\",\"v1_28\"]}}", "gzip-header_1.0.0": "{\"dependencies\":[{\"name\":\"crc32fast\",\"req\":\"^1.2.1\"}],\"features\":{}}", "h2_0.4.13": "{\"dependencies\":[{\"name\":\"atomic-waker\",\"req\":\"^1.0.0\"},{\"name\":\"bytes\",\"req\":\"^1\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"env_logger\",\"req\":\"^0.10\"},{\"name\":\"fnv\",\"req\":\"^1.0.5\"},{\"default_features\":false,\"name\":\"futures-core\",\"req\":\"^0.3\"},{\"default_features\":false,\"name\":\"futures-sink\",\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"hex\",\"req\":\"^0.4.3\"},{\"name\":\"http\",\"req\":\"^1\"},{\"features\":[\"std\"],\"name\":\"indexmap\",\"req\":\"^2\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"quickcheck\",\"req\":\"^1.0.3\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.8.4\"},{\"kind\":\"dev\",\"name\":\"serde\",\"req\":\"^1.0.0\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0.0\"},{\"name\":\"slab\",\"req\":\"^0.4.2\"},{\"features\":[\"io-util\"],\"name\":\"tokio\",\"req\":\"^1\"},{\"features\":[\"rt-multi-thread\",\"macros\",\"sync\",\"net\"],\"kind\":\"dev\",\"name\":\"tokio\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"tokio-rustls\",\"req\":\"^0.26\"},{\"features\":[\"codec\",\"io\"],\"name\":\"tokio-util\",\"req\":\"^0.7.1\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"tracing\",\"req\":\"^0.1.35\"},{\"kind\":\"dev\",\"name\":\"walkdir\",\"req\":\"^2.3.2\"},{\"kind\":\"dev\",\"name\":\"webpki-roots\",\"req\":\"^1\"}],\"features\":{\"stream\":[],\"unstable\":[]}}", - "h2_0.4.16": "{\"dependencies\":[{\"name\":\"atomic-waker\",\"req\":\"^1.0.0\"},{\"name\":\"bytes\",\"req\":\"^1\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"env_logger\",\"req\":\"^0.10\"},{\"name\":\"fnv\",\"req\":\"^1.0.5\"},{\"default_features\":false,\"name\":\"futures-core\",\"req\":\"^0.3\"},{\"default_features\":false,\"name\":\"futures-sink\",\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"hex\",\"req\":\"^0.4.3\"},{\"name\":\"http\",\"req\":\"^1.1\"},{\"features\":[\"std\"],\"name\":\"indexmap\",\"req\":\"^2\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"quickcheck\",\"req\":\"^1.0.3\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.8.4\"},{\"kind\":\"dev\",\"name\":\"serde\",\"req\":\"^1.0.0\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0.0\"},{\"name\":\"slab\",\"req\":\"^0.4.2\"},{\"features\":[\"io-util\"],\"name\":\"tokio\",\"req\":\"^1\"},{\"features\":[\"rt-multi-thread\",\"macros\",\"sync\",\"net\"],\"kind\":\"dev\",\"name\":\"tokio\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"tokio-rustls\",\"req\":\"^0.26\"},{\"features\":[\"codec\",\"io\"],\"name\":\"tokio-util\",\"req\":\"^0.7.1\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"tracing\",\"req\":\"^0.1.35\"},{\"kind\":\"dev\",\"name\":\"walkdir\",\"req\":\"^2.3.2\"},{\"kind\":\"dev\",\"name\":\"webpki-roots\",\"req\":\"^1\"}],\"features\":{\"stream\":[],\"unstable\":[]}}", + "h2_0.4.19": "{\"dependencies\":[{\"name\":\"atomic-waker\",\"req\":\"^1.0.0\"},{\"name\":\"bytes\",\"req\":\"^1\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"env_logger\",\"req\":\"^0.10\"},{\"name\":\"fnv\",\"req\":\"^1.0.5\"},{\"default_features\":false,\"name\":\"futures-core\",\"req\":\"^0.3\"},{\"default_features\":false,\"name\":\"futures-sink\",\"req\":\"^0.3\"},{\"kind\":\"dev\",\"name\":\"hex\",\"req\":\"^0.4.3\"},{\"name\":\"http\",\"req\":\"^1.1\"},{\"features\":[\"std\"],\"name\":\"indexmap\",\"req\":\"^2\"},{\"default_features\":false,\"kind\":\"dev\",\"name\":\"quickcheck\",\"req\":\"^1.0.3\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.8.4\"},{\"kind\":\"dev\",\"name\":\"serde\",\"req\":\"^1.0.0\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0.0\"},{\"name\":\"slab\",\"req\":\"^0.4.2\"},{\"features\":[\"io-util\"],\"name\":\"tokio\",\"req\":\"^1\"},{\"features\":[\"rt-multi-thread\",\"macros\",\"sync\",\"net\"],\"kind\":\"dev\",\"name\":\"tokio\",\"req\":\"^1\"},{\"kind\":\"dev\",\"name\":\"tokio-rustls\",\"req\":\"^0.26\"},{\"features\":[\"codec\",\"io\"],\"name\":\"tokio-util\",\"req\":\"^0.7.1\"},{\"default_features\":false,\"features\":[\"std\"],\"name\":\"tracing\",\"req\":\"^0.1.35\"},{\"kind\":\"dev\",\"name\":\"walkdir\",\"req\":\"^2.3.2\"},{\"kind\":\"dev\",\"name\":\"webpki-roots\",\"req\":\"^1\"}],\"features\":{\"stream\":[],\"unstable\":[]}}", "half_2.7.1": "{\"dependencies\":[{\"features\":[\"derive\"],\"name\":\"arbitrary\",\"optional\":true,\"req\":\"^1.4.1\"},{\"default_features\":false,\"features\":[\"derive\"],\"name\":\"bytemuck\",\"optional\":true,\"req\":\"^1.4.1\"},{\"name\":\"cfg-if\",\"req\":\"^1.0.0\"},{\"kind\":\"dev\",\"name\":\"criterion\",\"req\":\"^0.5\"},{\"name\":\"crunchy\",\"req\":\"^0.2.2\",\"target\":\"cfg(target_arch = \\\"spirv\\\")\"},{\"kind\":\"dev\",\"name\":\"crunchy\",\"req\":\"^0.2.2\"},{\"default_features\":false,\"features\":[\"libm\"],\"name\":\"num-traits\",\"optional\":true,\"req\":\"^0.2.16\"},{\"kind\":\"dev\",\"name\":\"quickcheck\",\"req\":\"^1.0\"},{\"kind\":\"dev\",\"name\":\"quickcheck_macros\",\"req\":\"^1.0\"},{\"default_features\":false,\"features\":[\"thread_rng\"],\"name\":\"rand\",\"optional\":true,\"req\":\"^0.9.0\"},{\"kind\":\"dev\",\"name\":\"rand\",\"req\":\"^0.9.0\"},{\"default_features\":false,\"name\":\"rand_distr\",\"optional\":true,\"req\":\"^0.5.0\"},{\"name\":\"rkyv\",\"optional\":true,\"req\":\"^0.8.0\"},{\"default_features\":false,\"features\":[\"derive\"],\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0\"},{\"default_features\":false,\"features\":[\"derive\",\"simd\"],\"name\":\"zerocopy\",\"req\":\"^0.8.26\"}],\"features\":{\"alloc\":[],\"default\":[\"std\"],\"nightly\":[],\"rand_distr\":[\"dep:rand\",\"dep:rand_distr\"],\"std\":[\"alloc\"],\"use-intrinsics\":[],\"zerocopy\":[]}}", "hash32_0.2.1": "{\"dependencies\":[{\"default_features\":false,\"name\":\"byteorder\",\"req\":\"^1.2.2\"},{\"kind\":\"dev\",\"name\":\"hash32-derive\",\"req\":\"^0.1.0\"}],\"features\":{}}", "hash32_0.3.1": "{\"dependencies\":[{\"default_features\":false,\"name\":\"byteorder\",\"req\":\"^1.2.2\"}],\"features\":{}}", diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 01a121bf7833..e0242af21da8 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -8478,9 +8478,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.16" +version = "0.4.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27" +checksum = "ef8e5e5a340588f4452631496976cf8636d4a7ecf600239fdc27615d2530bc16" dependencies = [ "atomic-waker", "bytes", diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index a00c76da30b4..ef310acf2cb7 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -37,6 +37,10 @@ use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; use crate::guardian_v2::GuardianV2Event; use crate::now_unix_millis; +use crate::product_attribution::MAX_THREAD_PRODUCTS; +use crate::product_attribution::ThreadProductUpdate; +use crate::product_attribution::ThreadProducts; +use crate::product_attribution::product_event_batches; use crate::reducer::AnalyticsReducer; use crate::reducer::MAX_PLUGIN_MEASUREMENTS_PER_BATCH; use crate::reducer::tracked_tool_item_id; @@ -74,6 +78,8 @@ use std::collections::HashSet; use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; use std::time::Duration; use tokio::sync::mpsc; use tokio::sync::oneshot; @@ -86,12 +92,18 @@ const ANALYTICS_EVENT_DEDUPE_MAX_KEYS: usize = 4096; pub(crate) enum AnalyticsEventsQueueMessage { Fact(Box), + ThreadProduct { + thread_id: String, + product_sku: Option, + generation: u64, + }, Flush(oneshot::Sender<()>), } #[derive(Clone)] pub(crate) struct AnalyticsEventsQueue { pub(crate) sender: mpsc::Sender, + pub(crate) product_generation: Arc, pub(crate) app_used_emitted_keys: Arc>>, pub(crate) plugin_used_emitted_keys: Arc>>, } @@ -159,26 +171,63 @@ fn analytics_capture_file_from_env() -> Option { impl AnalyticsEventsQueue { fn new(auth_manager: Arc, destination: AnalyticsEventsDestination) -> Self { let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE); + let product_generation = Arc::new(AtomicU64::new(0)); + let worker_product_generation = Arc::clone(&product_generation); tokio::spawn(async move { let mut reducer = AnalyticsReducer::default(); + let mut thread_products = ThreadProducts::default(); + let mut registered_generation = 0; while let Some(input) = receiver.recv().await { + let generation = worker_product_generation.load(Ordering::Acquire); + if generation != registered_generation { + // A dropped registration invalidates both current and queued attribution. + thread_products = ThreadProducts::default(); + registered_generation = generation; + } let input = match input { AnalyticsEventsQueueMessage::Fact(input) => *input, + AnalyticsEventsQueueMessage::ThreadProduct { + thread_id, + product_sku, + generation, + } => { + if generation != registered_generation { + continue; + } + // Pending tool events belong to the configuration that collected them. + let mut events = Vec::new(); + if thread_products.get(&thread_id) != product_sku.as_deref() { + reducer.flush_thread(&thread_id, &mut events); + } + if product_sku.is_some() + && !thread_products.products.contains_key(&thread_id) + && thread_products.products.len() >= MAX_THREAD_PRODUCTS + && let Some(evicted) = thread_products.registered_threads.front() + { + reducer.flush_thread(evicted, &mut events); + } + send_track_events(&auth_manager, &destination, events, &thread_products) + .await; + thread_products.register(thread_id, product_sku); + continue; + } AnalyticsEventsQueueMessage::Flush(done_tx) => { let mut events = Vec::new(); reducer.flush(&mut events); - send_track_events(&auth_manager, &destination, events).await; + send_track_events(&auth_manager, &destination, events, &thread_products) + .await; let _ = done_tx.send(()); continue; } }; let mut events = Vec::new(); reducer.ingest(input, &mut events).await; - send_track_events(&auth_manager, &destination, events).await; + send_track_events(&auth_manager, &destination, events, &thread_products).await; } }); Self { sender, + product_generation, app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), } @@ -238,6 +287,30 @@ impl AnalyticsEventsQueue { } impl AnalyticsEventsClient { + /// Registers this thread's product attribution without changing authentication. + /// A full queue clears shared attribution until threads register again, without blocking startup. + pub fn update_thread_product_sku(&self, thread_id: ThreadId, update: ThreadProductUpdate) { + let product_sku = match update { + ThreadProductUpdate::Set(product) => Some(product), + ThreadProductUpdate::Clear => None, + }; + let Some(queue) = &self.queue else { + return; + }; + if queue + .sender + .try_send(AnalyticsEventsQueueMessage::ThreadProduct { + thread_id: thread_id.to_string(), + product_sku, + generation: queue.product_generation.load(Ordering::Acquire), + }) + .is_err() + { + queue.product_generation.fetch_add(1, Ordering::AcqRel); + tracing::warn!("clearing analytics product attribution: queue is unavailable"); + } + } + pub fn new( auth_manager: Arc, base_url: String, @@ -861,6 +934,7 @@ async fn send_track_events( auth_manager: &AuthManager, destination: &AnalyticsEventsDestination, mut events: Vec, + thread_products: &ThreadProducts, ) { if events.is_empty() { return; @@ -879,8 +953,33 @@ async fn send_track_events( return; } - for events in track_event_request_batches(events) { - send_track_events_request(&auth, destination, events, &http_client_factory).await; + // Product streams are capped by product_event_batches. Parallel streams add no + // timeout waves to the existing sequential isolated-event request splitting. + let mut requests = tokio::task::JoinSet::new(); + for (product_sku, events) in product_event_batches(events, thread_products) { + let (auth, destination, factory) = ( + auth.clone(), + destination.clone(), + http_client_factory.clone(), + ); + let product_sku = product_sku.map(str::to_string); + requests.spawn(async move { + for events in track_event_request_batches(events) { + send_track_events_request( + &auth, + &destination, + events, + &factory, + product_sku.as_deref(), + ) + .await; + } + }); + } + while let Some(result) = requests.join_next().await { + if let Err(error) = result { + tracing::warn!(%error, "analytics product stream failed"); + } } } @@ -912,6 +1011,7 @@ async fn send_track_events_request( destination: &AnalyticsEventsDestination, events: Vec, http_client_factory: &codex_http_client::HttpClientFactory, + product_sku: Option<&str>, ) { if events.is_empty() { return; @@ -943,14 +1043,16 @@ async fn send_track_events_request( return; } }; - let response = client + let mut request = client .post(url) .timeout(ANALYTICS_EVENTS_TIMEOUT) .headers(codex_model_provider::auth_provider_from_auth(auth).to_auth_headers()) .header("Content-Type", "application/json") - .json(&payload) - .send() - .await; + .json(&payload); + if let Some(product_sku) = product_sku { + request = request.header("X-OpenAI-Product-Sku", product_sku); + } + let response = request.send().await; match response { Ok(response) if response.status().is_success() => {} @@ -974,6 +1076,9 @@ fn capture_track_events_request( return false; }; + use std::sync::PoisonError; + static CAPTURE_LOCK: Mutex<()> = Mutex::new(()); + let _capture_guard = CAPTURE_LOCK.lock().unwrap_or_else(PoisonError::into_inner); if let Err(err) = crate::analytics_capture::append_payload(path, payload) { tracing::error!( path = %path.display(), diff --git a/codex-rs/analytics/src/client_product_tests.rs b/codex-rs/analytics/src/client_product_tests.rs new file mode 100644 index 000000000000..029dea7fd103 --- /dev/null +++ b/codex-rs/analytics/src/client_product_tests.rs @@ -0,0 +1,345 @@ +//! Verify product attribution remains per thread across shared queues and HTTP batches. +//! Reconfiguration preserves buffered attribution; queue overflow falls back to unknown. + +use super::AnalyticsEventsClient; +use super::AnalyticsEventsDestination; +use super::AnalyticsEventsQueue; +use super::sample_regular_track_event; +use super::sample_thread_start_response; +use super::sample_turn_start_response; +use super::send_track_events; +use crate::ThreadProductUpdate; +use crate::events::AppServerRpcTransport; +use crate::facts::PluginMeasurementRow; +use crate::facts::PluginMeasurementsInput; +use crate::product_attribution::ThreadProducts; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientResponsePayload; +use codex_app_server_protocol::CollabAgentTool; +use codex_app_server_protocol::CollabAgentToolCallStatus; +use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::ItemCompletedNotification; +use codex_app_server_protocol::ItemStartedNotification; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::TurnStartedNotification; +use codex_login::AuthManager; +use codex_protocol::ThreadId; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; + +enum Responses { + Immediate, + PauseAfter(usize), + Together, +} + +type CapturedRequest = (Option, serde_json::Value); + +#[test] +fn product_registry_clears_unset_threads_and_bounds_retention() { + let mut products = ThreadProducts::default(); + products.register("unset".to_string(), Some("aeon".to_string())); + products.register("unset".to_string(), /*product*/ None); + for index in 0..=4096 { + products.register(format!("thread-{index}"), Some("tpp".to_string())); + } + let batches = crate::product_attribution::product_event_batches( + vec![ + sample_regular_track_event("unset"), + sample_regular_track_event("thread-0"), + sample_regular_track_event("thread-4096"), + ], + &products, + ); + assert_eq!( + batches + .into_iter() + .map(|(product, events)| (product, events.len())) + .collect::>(), + vec![(None, 2), (Some("tpp"), 1)] + ); + for index in 0..17 { + products.register(format!("thread-{index}"), Some(format!("product-{index}"))); + } + let events = (0..17) + .map(|index| sample_regular_track_event(&format!("thread-{index}"))) + .collect::>(); + let expected = serde_json::json!([[null, &events]]); + assert_eq!( + serde_json::to_value(crate::product_attribution::product_event_batches( + events, &products + )) + .unwrap(), + expected + ); +} + +async fn capture_requests( + count: usize, + responses: Responses, +) -> ( + AnalyticsEventsDestination, + tokio::task::JoinHandle>, + tokio::sync::oneshot::Receiver<()>, + tokio::sync::oneshot::Sender<()>, +) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let destination = AnalyticsEventsDestination::Http { + url: format!("http://{}/events", listener.local_addr().unwrap()), + }; + let (paused_tx, paused_rx) = tokio::sync::oneshot::channel(); + let (resume_tx, resume_rx) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + let mut pause = Some((paused_tx, resume_rx)); + let mut requests = Vec::new(); + let mut pending = Vec::new(); + for _ in 0..count { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let (headers, body) = loop { + let mut buffer = [0; 4096]; + let count = stream.read(&mut buffer).await.unwrap(); + assert!(count > 0); + request.extend_from_slice(&buffer[..count]); + if let Some(end) = request.windows(4).position(|part| part == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&request[..end]).to_lowercase(); + let length: usize = headers + .lines() + .find_map(|line| line.strip_prefix("content-length: ")) + .unwrap() + .parse() + .unwrap(); + if request.len() >= end + 4 + length { + break ( + headers, + serde_json::from_slice(&request[end + 4..end + 4 + length]).unwrap(), + ); + } + } + }; + let product = headers + .lines() + .find_map(|line| line.strip_prefix("x-openai-product-sku: ")) + .map(str::to_string); + requests.push((product, body)); + if matches!(responses, Responses::PauseAfter(count) if count == requests.len()) { + let (paused, resume) = pause.take().unwrap(); + paused.send(()).unwrap(); + resume.await.unwrap(); + } + pending.push(stream); + if matches!(responses, Responses::Together) && requests.len() < count { + continue; + } + for mut stream in pending.drain(..) { + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + } + } + requests + }); + (destination, server, paused_rx, resume_tx) +} + +#[tokio::test] +async fn product_streams_send_concurrently_and_preserve_event_order() { + // Withhold every response until all 16 requests arrive: serial delivery would time out. + let (destination, server, _, _) = capture_requests(/*count*/ 16, Responses::Together).await; + let manager = AuthManager::from_auth_for_testing( + codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(), + ); + let mut products = ThreadProducts::default(); + for index in 0..16 { + products.register(format!("thread-{index}"), Some(format!("product-{index}"))); + } + let event = |index| { + let mut event = sample_regular_track_event(&format!("thread-{}", index % 16)); + if let crate::events::TrackEventRequest::SkillInvocation(skill) = &mut event { + skill.event_params.turn_id = Some(format!("turn-{index}")); + } + event + }; + let events = (0..32).map(event).collect(); + tokio::time::timeout( + std::time::Duration::from_secs(15), + send_track_events(&manager, &destination, events, &products), + ) + .await + .unwrap(); + let mut requests = server.await.unwrap(); + requests.sort_by(|a, b| a.0.cmp(&b.0)); + let mut expected = (0..16) + .map(|index| { + ( + Some(format!("product-{index}")), + serde_json::json!({"events": [event(index), event(index + 16)]}), + ) + }) + .collect::>(); + expected.sort_by(|a, b| a.0.cmp(&b.0)); + assert_eq!(requests, expected); +} + +#[tokio::test] +async fn buffered_tool_events_preserve_attribution_or_drop_it_on_queue_overflow() { + for (new_product, overflow) in [ + (Some("tpp"), false), + (None, false), + (Some("tpp"), true), + (None, true), + ] { + let (destination, server, paused, resume) = capture_requests( + if overflow { 5 } else { 2 }, + if overflow { + Responses::PauseAfter(2) + } else { + Responses::Immediate + }, + ) + .await; + let manager = AuthManager::from_auth_for_testing( + codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(), + ); + let client = AnalyticsEventsClient { + queue: Some(AnalyticsEventsQueue::new(manager, destination)), + }; + let thread_id = ThreadId::new(); + client.update_thread_product_sku(thread_id, ThreadProductUpdate::Set("aeon".to_string())); + client.track_initialize( + /*connection_id*/ 1, + InitializeParams { + client_info: ClientInfo { + name: "test".to_string(), + title: None, + version: "1".to_string(), + }, + capabilities: None, + }, + "test_client".to_string(), + AppServerRpcTransport::InProcess, + ); + let mut response = sample_thread_start_response(); + let ClientResponsePayload::ThreadStart(started) = &mut response else { + unreachable!() + }; + started.thread.id = thread_id.to_string(); + client.track_response(/*connection_id*/ 1, RequestId::Integer(1), &response); + let ClientResponsePayload::TurnStart(started) = sample_turn_start_response() else { + unreachable!() + }; + client.track_notification(&ServerNotification::TurnStarted(TurnStartedNotification { + thread_id: thread_id.to_string(), + turn: started.turn, + })); + // Collaborator completions wait in the reducer for subsequent sampling evidence. + let completed = ItemCompletedNotification { + thread_id: thread_id.to_string(), + turn_id: "turn-1".to_string(), + completed_at_ms: 2, + item: ThreadItem::CollabAgentToolCall { + id: "item".to_string(), + tool: CollabAgentTool::SendMessage, + status: CollabAgentToolCallStatus::Failed, + sender_thread_id: thread_id.to_string(), + receiver_thread_ids: Vec::new(), + prompt: None, + model: None, + reasoning_effort: None, + agents_states: Default::default(), + }, + }; + client.track_notification(&ServerNotification::ItemStarted(ItemStartedNotification { + thread_id: thread_id.to_string(), + turn_id: "turn-1".to_string(), + started_at_ms: 1, + item: completed.item.clone(), + })); + client.track_notification(&ServerNotification::ItemCompleted(completed)); + let measurement = PluginMeasurementsInput { + thread_id: thread_id.to_string(), + turn_id: "turn-1".to_string(), + item_id: "measurement".to_string(), + originator: "test_client".to_string(), + model_slug: None, + reasoning_effort: None, + plugin_id: "sites@test".to_string(), + execution_id: "execution".to_string(), + operation: "build".to_string(), + rows: vec![PluginMeasurementRow { + measurement_name: "duration_ms".to_string(), + number_value: 12.0, + dimensions: BTreeMap::new(), + }], + }; + if overflow { + client.track_plugin_measurements(measurement.clone()); + paused.await.unwrap(); + // Hold HTTP delivery while stale registrations fill the shared queue. + let capacity = client.queue.as_ref().unwrap().sender.capacity(); + for _ in 0..capacity { + client.update_thread_product_sku( + thread_id, + ThreadProductUpdate::Set("aeon".to_string()), + ); + } + } + client.update_thread_product_sku( + thread_id, + new_product.map_or(ThreadProductUpdate::Clear, |value| { + ThreadProductUpdate::Set(value.to_string()) + }), + ); + if overflow { + resume.send(()).unwrap(); + } + client.flush().await; + if overflow { + client.track_plugin_measurements(measurement.clone()); + client.flush().await; + client + .update_thread_product_sku(thread_id, ThreadProductUpdate::Set("tpp".to_string())); + client.track_plugin_measurements(measurement); + client.flush().await; + } + let requests = tokio::time::timeout(std::time::Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(); + let expected = if overflow { + vec![ + (Some("aeon"), "codex_thread_initialized"), + (Some("aeon"), "codex_plugin_measurement_event"), + (None, "codex_collab_agent_tool_call_event"), + (None, "codex_plugin_measurement_event"), + (Some("tpp"), "codex_plugin_measurement_event"), + ] + } else { + vec![ + (Some("aeon"), "codex_thread_initialized"), + (Some("aeon"), "codex_collab_agent_tool_call_event"), + ] + } + .into_iter() + .map(|(product, event)| (product.map(str::to_string), event.to_string())) + .collect::>(); + assert_eq!( + requests + .into_iter() + .map(|(product, body)| ( + product, + body["events"][0]["event_type"] + .as_str() + .unwrap() + .to_string(), + )) + .collect::>(), + expected + ); + } +} diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 9112070c9c47..b4fdf1066b5e 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -4,7 +4,6 @@ use super::AnalyticsEventsQueue; use super::AnalyticsEventsQueueMessage; #[cfg(debug_assertions)] use super::capture_track_events_request; -#[cfg(debug_assertions)] use super::send_track_events; #[cfg(debug_assertions)] use super::send_track_events_request; @@ -60,6 +59,8 @@ use crate::facts::InvocationType; use crate::facts::PluginMeasurementRow; use crate::facts::PluginMeasurementsInput; use crate::facts::TrackEventsContext; +#[cfg(debug_assertions)] +use crate::product_attribution::ThreadProducts; use crate::reducer::MAX_PLUGIN_MEASUREMENTS_PER_BATCH; use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; @@ -108,6 +109,9 @@ use std::time::SystemTime; use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; +#[path = "client_product_tests.rs"] +mod product_tests; + #[cfg(debug_assertions)] impl AnalyticsEventsClient { pub(crate) fn new_for_capture_file(auth_manager: Arc, path: PathBuf) -> Self { @@ -289,6 +293,7 @@ fn client_with_receiver() -> ( let (sender, receiver) = mpsc::channel(8); let queue = AnalyticsEventsQueue { sender, + product_generation: Arc::new(std::sync::atomic::AtomicU64::new(0)), app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), }; @@ -370,7 +375,14 @@ async fn capture_file_writes_exact_serialized_request() { let auth = codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(); let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault); - send_track_events_request(&auth, &destination, vec![event], &factory).await; + send_track_events_request( + &auth, + &destination, + vec![event], + &factory, + /*product_sku*/ None, + ) + .await; let contents = fs::read_to_string(&capture_path).expect("read capture file"); let lines = contents.lines().collect::>(); @@ -398,7 +410,14 @@ async fn capture_file_writes_final_batches_as_separate_lines() { for batch in track_event_request_batches(events) { let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault); - send_track_events_request(&auth, &destination, batch, &factory).await; + send_track_events_request( + &auth, + &destination, + batch, + &factory, + /*product_sku*/ None, + ) + .await; } let contents = fs::read_to_string(&capture_path).expect("read capture file"); @@ -474,6 +493,7 @@ async fn api_key_auth_sends_only_plugin_events_to_codex_backend() { sample_artifact_operation_event("plugin-artifact"), plugin_measurement("plugin-measurement", "sample@test"), ], + &ThreadProducts::default(), ) .await; diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index f98e3ad9b406..6160dd704a0a 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -5,6 +5,7 @@ mod client; mod events; mod facts; mod guardian_v2; +mod product_attribution; mod reducer; mod thread_hint; @@ -78,6 +79,7 @@ pub use facts::TurnTokenUsageFact; pub use facts::build_track_events_context; pub use guardian_v2::GuardianV2Event; pub use guardian_v2::GuardianV2EventKind; +pub use product_attribution::ThreadProductUpdate; pub use thread_hint::ThreadHintStatus; pub use thread_hint::ThreadHintStatusEvent; diff --git a/codex-rs/analytics/src/product_attribution.rs b/codex-rs/analytics/src/product_attribution.rs new file mode 100644 index 000000000000..4fa872102e68 --- /dev/null +++ b/codex-rs/analytics/src/product_attribution.rs @@ -0,0 +1,109 @@ +//! Keep per-thread product attribution separate from the shared analytics identity. + +use crate::events::TrackEventRequest; +use std::collections::HashMap; +use std::collections::VecDeque; + +pub(crate) const MAX_THREAD_PRODUCTS: usize = 4096; +const MAX_PRODUCT_BATCHES: usize = 16; + +/// An explicit change to a thread's product attribution, independent of authentication. +#[derive(Clone)] +pub enum ThreadProductUpdate { + Set(String), + Clear, +} + +#[derive(Default)] +pub(crate) struct ThreadProducts { + pub(crate) products: HashMap, + pub(crate) registered_threads: VecDeque, +} + +impl ThreadProducts { + pub(crate) fn get(&self, thread_id: &str) -> Option<&str> { + self.products.get(thread_id).map(String::as_str) + } + + pub(crate) fn register(&mut self, thread_id: String, product: Option) { + self.registered_threads.retain(|id| id != &thread_id); + self.products.remove(&thread_id); + let Some(product) = product else { + return; + }; + // Attribution is best-effort. Evicted threads have no product header. + if self.products.len() >= MAX_THREAD_PRODUCTS + && let Some(oldest) = self.registered_threads.pop_front() + { + self.products.remove(&oldest); + } + self.registered_threads.push_back(thread_id.clone()); + self.products.insert(thread_id, product); + } +} + +/// Preserve order within each product, with a bounded number of concurrent streams. +/// Too many products drops only attribution, preserving the original event order. +pub(crate) fn product_event_batches( + events: Vec, + thread_products: &ThreadProducts, +) -> Vec<(Option<&str>, Vec)> { + let products: std::collections::HashSet<_> = events + .iter() + .map(|event| event_thread_id(event).and_then(|id| thread_products.get(id))) + .collect(); + if products.len() > MAX_PRODUCT_BATCHES { + return vec![(None, events)]; + } + let mut batches: Vec<(Option<&str>, Vec)> = Vec::new(); + for event in events { + let product = event_thread_id(&event).and_then(|thread_id| thread_products.get(thread_id)); + if let Some((_, batch)) = batches.iter_mut().find(|(sku, _)| *sku == product) { + batch.push(event); + } else { + batches.push((product, vec![event])); + } + } + batches +} + +fn event_thread_id(event: &TrackEventRequest) -> Option<&str> { + match event { + TrackEventRequest::SkillInvocation(event) => event.event_params.thread_id.as_deref(), + TrackEventRequest::ThreadInitialized(event) => Some(&event.event_params.thread_id), + TrackEventRequest::ThreadArchive(event) => Some(&event.event_params.thread_id), + TrackEventRequest::GuardianReview(event) => { + Some(&event.event_params.guardian_review.thread_id) + } + TrackEventRequest::GuardianV2(event) => Some(&event.event_params.guardian_v2.thread_id), + TrackEventRequest::AppMentioned(event) => event.event_params.thread_id.as_deref(), + TrackEventRequest::AppUsed(event) => event.event_params.app.thread_id.as_deref(), + TrackEventRequest::HookRun(event) => event.event_params.thread_id.as_deref(), + TrackEventRequest::Compaction(event) => Some(&event.event_params.thread_id), + TrackEventRequest::Goal(event) => Some(&event.event_params.thread_id), + TrackEventRequest::ThreadHintStatus(event) => Some(&event.event_params.thread_id), + TrackEventRequest::TurnEvent(event) => Some(&event.event_params.thread_id), + TrackEventRequest::TurnSteer(event) => Some(&event.event_params.thread_id), + TrackEventRequest::ArtifactOperation(event) => Some(&event.event_params.thread_id), + TrackEventRequest::CommandExecution(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::PluginMeasurement(event) => Some(&event.event_params.thread_id), + TrackEventRequest::FileChange(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::McpToolCall(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::DynamicToolCall(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::ControlToolCall(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::CollabAgentToolCall(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::WebSearch(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::ImageGeneration(event) => Some(&event.event_params.base.thread_id), + TrackEventRequest::AcceptedLineFingerprints(event) => Some(&event.event_params.thread_id), + TrackEventRequest::ReviewEvent(event) => Some(&event.event_params.thread_id), + TrackEventRequest::PluginUsed(event) => event.event_params.thread_id.as_deref(), + TrackEventRequest::PluginInstallRequested(event) => Some(&event.event_params.thread_id), + TrackEventRequest::PluginInstalled(_) + | TrackEventRequest::PluginUninstalled(_) + | TrackEventRequest::PluginEnabled(_) + | TrackEventRequest::PluginDisabled(_) + | TrackEventRequest::PluginInstallFailed(_) + | TrackEventRequest::ExternalAgentConfigImportCompleted(_) + | TrackEventRequest::ExternalAgentConfigImportFailure(_) => None, + } +} diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index d9a00901888e..7640223e429f 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -157,7 +157,7 @@ use codex_app_server_protocol::TurnSteerResponse; use codex_app_server_protocol::UserInput; use codex_app_server_protocol::WebSearchAction; use codex_git_utils::SanitizedGitUrl; -use codex_git_utils::collect_git_info; +use codex_git_utils::get_git_origin_url; use codex_git_utils::get_git_repo_root; use codex_login::default_client::originator; use codex_protocol::config_types::ModeKind; @@ -1155,6 +1155,14 @@ impl AnalyticsReducer { } } + pub(crate) fn flush_thread(&mut self, thread_id: &str, out: &mut Vec) { + for ((pending_thread_id, _), state) in &mut self.tool_response_states { + if pending_thread_id == thread_id { + out.extend(state.pending_tool_events.drain(..)); + } + } + } + fn ingest_initialize( &mut self, connection_id: u64, @@ -1348,9 +1356,7 @@ impl AnalyticsReducer { }; let repo_root = get_git_repo_root(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) + get_git_origin_url(root).await } else { None }; diff --git a/codex-rs/analytics/src/tests/suite/plugins_tests.rs b/codex-rs/analytics/src/tests/suite/plugins_tests.rs index 268a36f3ada1..584f5d0c84d6 100644 --- a/codex-rs/analytics/src/tests/suite/plugins_tests.rs +++ b/codex-rs/analytics/src/tests/suite/plugins_tests.rs @@ -286,6 +286,7 @@ fn plugin_used_dedupe_is_keyed_by_turn_and_plugin() { let (sender, _receiver) = mpsc::channel(1); let queue = AnalyticsEventsQueue { sender, + product_generation: Arc::new(std::sync::atomic::AtomicU64::new(0)), app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), }; diff --git a/codex-rs/app-server/src/thread_status.rs b/codex-rs/app-server/src/thread_status.rs index 4f0a83e7056b..c627db4b7f0e 100644 --- a/codex-rs/app-server/src/thread_status.rs +++ b/codex-rs/app-server/src/thread_status.rs @@ -227,11 +227,7 @@ impl ThreadWatchManager { let notification = { let mut state = self.state.lock().await; let notification = mutate(&mut state); - let running_turn_count = state - .runtime_by_thread_id - .values() - .filter(|runtime| runtime.running) - .count(); + let running_turn_count = state.running_turn_count; self.running_turn_count_tx.send_if_modified(|current| { if *current == running_turn_count { false @@ -311,6 +307,7 @@ pub(crate) fn resolve_thread_status( struct ThreadWatchState { runtime_by_thread_id: HashMap, status_watcher_by_thread_id: HashMap>, + running_turn_count: usize, } impl ThreadWatchState { @@ -335,7 +332,13 @@ impl ThreadWatchState { fn remove_thread(&mut self, thread_id: &str) -> Option { let previous_status = self.status_for(thread_id); - self.runtime_by_thread_id.remove(thread_id); + if self + .runtime_by_thread_id + .remove(thread_id) + .is_some_and(|runtime| runtime.running) + { + self.running_turn_count -= 1; + } self.update_status_watcher(thread_id, &ThreadStatus::NotLoaded); if previous_status.is_some() && previous_status != Some(ThreadStatus::NotLoaded) { Some(ThreadStatusChangedNotification { @@ -360,8 +363,14 @@ impl ThreadWatchState { .runtime_by_thread_id .entry(thread_id.to_string()) .or_default(); + let was_running = runtime.running; runtime.is_loaded = true; mutate(runtime); + match (was_running, runtime.running) { + (false, true) => self.running_turn_count += 1, + (true, false) => self.running_turn_count -= 1, + (false, false) | (true, true) => {} + } self.update_status_watcher_for_thread(thread_id); self.status_changed_notification(thread_id.to_string(), previous_status) } @@ -712,11 +721,24 @@ mod tests { assert!(count.has_changed().expect("watch remains open")); assert_eq!(*count.borrow_and_update(), 1); + manager.note_turn_started(INTERACTIVE_THREAD_ID).await; + assert!(!count.has_changed().expect("watch remains open")); + + manager.note_turn_started(NON_INTERACTIVE_THREAD_ID).await; + assert!(count.has_changed().expect("watch remains open")); + assert_eq!(*count.borrow_and_update(), 2); + assert_eq!(manager.running_turn_count().await, 2); + let _permission_guard = manager .note_permission_requested(INTERACTIVE_THREAD_ID) .await; assert!(!count.has_changed().expect("watch remains open")); + manager.remove_thread(NON_INTERACTIVE_THREAD_ID).await; + assert!(count.has_changed().expect("watch remains open")); + assert_eq!(*count.borrow_and_update(), 1); + assert_eq!(manager.running_turn_count().await, 1); + manager.note_thread_shutdown(INTERACTIVE_THREAD_ID).await; assert!(count.has_changed().expect("watch remains open")); assert_eq!(*count.borrow_and_update(), 0); diff --git a/codex-rs/app-server/tests/common/models_cache.rs b/codex-rs/app-server/tests/common/models_cache.rs index 0b14bc00353c..8110221b6eb8 100644 --- a/codex-rs/app-server/tests/common/models_cache.rs +++ b/codex-rs/app-server/tests/common/models_cache.rs @@ -31,6 +31,7 @@ fn preset_to_info(preset: &ModelPreset, priority: i32) -> ModelInfo { available_access_programs: preset.available_access_programs.clone(), upgrade: preset.upgrade.as_ref().map(Into::into), model_messages: Some(ModelMessages { + content_filter_guidance: None, persistent_instructions: None, tools: None, instructions_template: Some("base instructions".to_string()), diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 51f3e06b9a54..d24896903618 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -89,6 +89,7 @@ mod permission_profile_list; mod plan_item; mod plugin_install; mod plugin_list; +mod plugin_manifest_cache; mod plugin_read; mod plugin_reconcile; mod plugin_search; diff --git a/codex-rs/app-server/tests/suite/v2/plugin_manifest_cache.rs b/codex-rs/app-server/tests/suite/v2/plugin_manifest_cache.rs new file mode 100644 index 000000000000..854e8a533c0a --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/plugin_manifest_cache.rs @@ -0,0 +1,95 @@ +//! Exercise manifest and overlay invalidation across RPCs in one running app-server. + +use anyhow::Result; +use app_test_support::TestAppServer; +use codex_app_server_protocol::PluginListMarketplaceKind; +use codex_app_server_protocol::PluginListParams; +use codex_app_server_protocol::PluginListResponse; +use codex_utils_absolute_path::AbsolutePathBuf; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::time::Duration; +use tempfile::TempDir; +use tokio::time::timeout; + +#[tokio::test] +async fn plugin_list_observes_manifest_and_overlay_revisions_in_one_server() -> Result<()> { + let codex_home = TempDir::new()?; + let repository = TempDir::new()?; + let root = repository.path(); + for directory in [".git", ".agents/plugins", "sample/.codex-plugin"] { + std::fs::create_dir_all(root.join(directory))?; + } + std::fs::write( + codex_home.path().join("config.toml"), + "[features]\nplugins = true\nremote_plugin = false\n", + )?; + std::fs::write( + root.join(".agents/plugins/marketplace.json"), + serde_json::to_vec(&json!({ + "name": "cache-test", "plugins": [{"name": "sample", "source": {"source": "local", "path": "./sample"}}] + }))?, + )?; + let mut server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized_with_timeout(Duration::from_secs(30)) + .await?; + let overlay_path = root.join("sample/.codex-plugin/plugin.json"); + for (version, overlay_name, expected_name) in [ + ("1.0.0", Some("First"), "First"), + ("2.0.0", Some("First"), "First"), + ("2.0.0", Some("Changed"), "Changed"), + ("2.0.0", None, "sample"), + ] { + std::fs::write( + root.join("sample/plugin.json"), + serde_json::to_vec(&json!({ + "$schema": "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json", + "name": "sample", "version": version + }))?, + )?; + if let Some(display_name) = overlay_name { + std::fs::write( + &overlay_path, + serde_json::to_vec(&json!({"interface": {"displayName": display_name}}))?, + )?; + } else { + std::fs::remove_file(&overlay_path)?; + } + let mut previous = None; + for _ in 0..2 { + let id = server + .send_plugin_list_request(PluginListParams { + cwds: Some(vec![AbsolutePathBuf::try_from(root)?]), + marketplace_kinds: Some(vec![PluginListMarketplaceKind::Local]), + force_refetch: false, + }) + .await?; + let response: PluginListResponse = + timeout(Duration::from_secs(30), server.read_response(id)).await??; + assert_eq!(response.marketplace_load_errors, Vec::new()); + let marketplace = response + .marketplaces + .iter() + .find(|entry| entry.name == "cache-test") + .expect("local marketplace"); + let plugin = &marketplace.plugins[0]; + assert_eq!( + ( + plugin.id.as_str(), + plugin.local_version.as_deref(), + plugin + .interface + .as_ref() + .and_then(|value| value.display_name.as_deref()) + ), + ("sample@cache-test", Some(version), Some(expected_name)) + ); + if let Some(previous) = previous { + assert_eq!(marketplace, &previous); + } + previous = Some(marketplace.clone()); + } + } + Ok(()) +} diff --git a/codex-rs/cli/src/doctor.rs b/codex-rs/cli/src/doctor.rs index d7014312c575..3dc3d07b835f 100644 --- a/codex-rs/cli/src/doctor.rs +++ b/codex-rs/cli/src/doctor.rs @@ -2429,7 +2429,8 @@ fn websocket_error_detail(err: &ApiError) -> String { format!("handshake API error: {status} {message}") } ApiError::Stream(message) => format!("handshake stream error: {message}"), - ApiError::ContextWindowExceeded + ApiError::ContentFilter + | ApiError::ContextWindowExceeded | ApiError::QuotaExceeded | ApiError::UsageNotIncluded | ApiError::Retryable { .. } diff --git a/codex-rs/codex-api/src/api_bridge.rs b/codex-rs/codex-api/src/api_bridge.rs index 73bc7a787b07..abf0bf0d7041 100644 --- a/codex-rs/codex-api/src/api_bridge.rs +++ b/codex-rs/codex-api/src/api_bridge.rs @@ -28,6 +28,7 @@ pub fn map_api_error(err: ApiError) -> CodexErr { ApiError::Transport(_) | ApiError::Api { .. } | ApiError::Stream(_) + | ApiError::ContentFilter | ApiError::ContextWindowExceeded | ApiError::QuotaExceeded | ApiError::UsageNotIncluded @@ -56,6 +57,7 @@ fn map_api_error_details(err: ApiError) -> CodexErr { CodexErr::new(CodexErrorDetails::RateLimitExceeded(message)) } ApiError::Stream(msg) => CodexErr::Stream(msg), + ApiError::ContentFilter => CodexErr::new(CodexErrorDetails::ContentFilter), ApiError::ServerOverloaded { .. } => CodexErr::ServerOverloaded, ApiError::FlexUnavailable => CodexErr::new(CodexErrorDetails::FlexUnavailable), ApiError::Api { status, message } => { diff --git a/codex-rs/codex-api/src/api_bridge_tests.rs b/codex-rs/codex-api/src/api_bridge_tests.rs index 252d2f2ed819..56c4838ef670 100644 --- a/codex-rs/codex-api/src/api_bridge_tests.rs +++ b/codex-rs/codex-api/src/api_bridge_tests.rs @@ -11,6 +11,25 @@ fn map_api_error_maps_server_overloaded() { assert!(matches!(err.details(), CodexErrorDetails::ServerOverloaded)); } +#[test] +fn map_api_error_preserves_content_filter_retry_and_public_error() { + let err = map_api_error(ApiError::ContentFilter); + assert!(matches!(err.details(), CodexErrorDetails::ContentFilter)); + assert_eq!( + ( + err.retry_delay(/*retry_count*/ 1).is_some(), + err.to_codex_protocol_error(), + err.to_string(), + ), + ( + true, + CodexErrorInfo::Other, + "stream disconnected before completion: Incomplete response returned, reason: content_filter" + .to_string(), + ) + ); +} + #[tokio::test(start_paused = true)] async fn map_api_error_preserves_retry_delay() { let retry_delay = std::time::Duration::from_secs(17); diff --git a/codex-rs/codex-api/src/error.rs b/codex-rs/codex-api/src/error.rs index e2c934ce8ed5..2834137129e7 100644 --- a/codex-rs/codex-api/src/error.rs +++ b/codex-rs/codex-api/src/error.rs @@ -14,6 +14,8 @@ pub enum ApiError { Api { status: StatusCode, message: String }, #[error("stream error: {0}")] Stream(String), + #[error("stream error: Incomplete response returned, reason: content_filter")] + ContentFilter, #[error("context window exceeded")] ContextWindowExceeded, #[error("quota exceeded")] diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index c01b13f1fe63..6b2a7ed47cb6 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -425,6 +425,9 @@ pub fn process_responses_event( .and_then(Value::as_str) }); let reason = reason.unwrap_or("unknown"); + if reason == "content_filter" { + return Err(ResponsesEventError::Api(ApiError::ContentFilter)); + } if reason != "interrupted" { let message = format!("Incomplete response returned, reason: {reason}"); return Err(ResponsesEventError::Api(ApiError::Stream(message))); diff --git a/codex-rs/core-plugins/src/discoverable_tests.rs b/codex-rs/core-plugins/src/discoverable_tests.rs index 3c2085c9e13f..df227071fda3 100644 --- a/codex-rs/core-plugins/src/discoverable_tests.rs +++ b/codex-rs/core-plugins/src/discoverable_tests.rs @@ -640,7 +640,7 @@ async fn omits_not_available_curated_plugins() { } #[tokio::test] -async fn does_not_reload_marketplace_per_plugin() { +async fn discovery_warns_once_per_invalid_manifest() { let codex_home = tempdir().expect("tempdir should succeed"); let curated_root = curated_plugins_repo_path(codex_home.path()); write_openai_curated_marketplace(&curated_root, &["slack", "gmail", "openai-developers"]); @@ -697,12 +697,12 @@ async fn does_not_reload_marketplace_per_plugin() { let logs = String::from_utf8(buffer.lock().expect("buffer lock").clone()) .expect("utf8 logs") .replace('\\', "/"); - assert_eq!(logs.matches("ignoring interface.defaultPrompt").count(), 8); - assert_eq!(logs.matches("gmail/.codex-plugin/plugin.json").count(), 4); + assert_eq!(logs.matches("ignoring interface.defaultPrompt").count(), 2); + assert_eq!(logs.matches("gmail/.codex-plugin/plugin.json").count(), 1); assert_eq!( logs.matches("openai-developers/.codex-plugin/plugin.json") .count(), - 4 + 1 ); } diff --git a/codex-rs/core-plugins/src/loader.rs b/codex-rs/core-plugins/src/loader.rs index d9f49eec4bf4..92500be8cc96 100644 --- a/codex-rs/core-plugins/src/loader.rs +++ b/codex-rs/core-plugins/src/loader.rs @@ -2,6 +2,7 @@ use crate::PluginGitMode; use crate::app_mcp_routing::apply_app_mcp_routing_policy; use crate::app_mcp_routing::apps_route_available; use crate::is_openai_curated_marketplace_name; +use crate::manifest::ManifestCache; use crate::manifest::PluginManifest; use crate::manifest::PluginManifestFormat; use crate::manifest::PluginManifestHooks; @@ -889,7 +890,7 @@ async fn load_plugin( return loaded_plugin; } - let Some(loaded_manifest) = load_plugin_manifest_with_format(plugin_root.as_path()) else { + let Some(loaded_manifest) = store.manifest_cache.load(plugin_root.as_path()) else { loaded_plugin.error = Some("missing or invalid plugin.json".to_string()); return loaded_plugin; }; @@ -942,7 +943,8 @@ async fn load_plugin( ) .await; if loaded_manifest.format == PluginManifestFormat::Legacy { - loaded_plugin.apps = load_plugin_apps(plugin_root.as_path()).await; + loaded_plugin.apps = + load_plugin_apps_from_manifest(plugin_root.as_path(), manifest_paths).await; } } PluginLoadScope::HooksOnly => {} @@ -1328,12 +1330,13 @@ async fn load_apps_from_paths( app_declarations } -pub async fn plugin_capability_summary_from_root( +pub(crate) async fn plugin_capability_summary_from_root( plugin_id: &PluginId, plugin_root: &AbsolutePathBuf, skill_root_loader: &dyn SkillRootLoader, + manifest_cache: &ManifestCache, ) -> Option { - let loaded_manifest = load_plugin_manifest_with_format(plugin_root.as_path())?; + let loaded_manifest = manifest_cache.load(plugin_root.as_path())?; let manifest_format = loaded_manifest.format; let manifest = loaded_manifest.manifest; let plugin_identity = PluginIdentity { diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 3312b401202b..e5614b9f55bc 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -14,6 +14,7 @@ use super::LoadedPlugin; use super::PluginLoadOutcome; use crate::PluginGitMode; use crate::app_mcp_routing::apply_app_mcp_routing_policy; +use crate::http_client_selector::HttpClientSelector; use crate::installed_marketplaces::installed_marketplace_roots_from_layer_stack; use crate::is_openai_curated_marketplace_name; use crate::loaded_cache_metrics; @@ -38,8 +39,6 @@ use crate::loader::refresh_non_curated_plugin_cache_force_reinstall_detailed; use crate::loader::remote_installed_plugins_to_config; use crate::manifest::PluginManifestFormat; use crate::manifest::PluginManifestInterface; -use crate::manifest::load_plugin_manifest; -use crate::manifest::load_plugin_manifest_with_format; use crate::marketplace::MarketplaceError; use crate::marketplace::MarketplaceInterface; use crate::marketplace::MarketplaceListError; @@ -52,7 +51,7 @@ use crate::marketplace::ResolvedMarketplacePlugin; use crate::marketplace::find_installable_marketplace_plugin; use crate::marketplace::find_marketplace_plugin; use crate::marketplace::home_dir; -use crate::marketplace::list_marketplaces_with_home; +use crate::marketplace::list_marketplaces_with_cache; use crate::marketplace::plugin_interface_with_marketplace_category; use crate::marketplace_policy::MarketplacePolicy; use crate::marketplace_policy::configured_plugins_from_stack; @@ -60,6 +59,7 @@ use crate::marketplace_upgrade::ConfigLayerReload; use crate::marketplace_upgrade::ConfiguredMarketplaceUpgradeError; use crate::marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome; use crate::marketplace_upgrade::upgrade_configured_git_marketplaces_with_mode; +use crate::remote::CODEX_PRODUCT_SKU; use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME; use crate::remote::RecommendedPluginsMode; use crate::remote::RemoteInstalledPlugin; @@ -100,7 +100,9 @@ use codex_config::types::ToolSuggestDiscoverableType; use codex_connectors::ConnectorSnapshot; use codex_connectors::PluginConnectorSource; use codex_hooks::plugin_hook_declarations; +use codex_http_client::ClientRouteClass; use codex_http_client::HttpClientFactory; +use codex_http_client::RouteAwareClientPool; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_plugin::AppConnectorId; @@ -130,6 +132,7 @@ use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; +use std::sync::OnceLock; use std::sync::RwLock; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; @@ -159,6 +162,7 @@ pub struct PluginsConfigInput { pub chatgpt_base_url: String, pub product_sku: Option, http_client_factory: HttpClientFactory, + remote_http_clients: Arc>>, } impl PluginsConfigInput { @@ -179,16 +183,29 @@ impl PluginsConfigInput { chatgpt_base_url, http_client_factory, product_sku, + remote_http_clients: Arc::new(OnceLock::new()), } } - /// Builds route-aware service state for remote plugin requests. + /// Shares a lazy route-aware connection pool across this input and its clones. + /// Request metadata remains current; new config inputs own a new pool. pub fn remote_plugin_service_config(&self) -> RemotePluginServiceConfig { - RemotePluginServiceConfig::new( - self.chatgpt_base_url.clone(), - self.http_client_factory.clone(), - self.product_sku.clone(), - ) + let http_clients = self.remote_http_clients.get_or_init(|| { + Arc::new( + RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging( + self.http_client_factory.clone(), + ClientRouteClass::Api, + ), + ) + }); + RemotePluginServiceConfig { + chatgpt_base_url: self.chatgpt_base_url.clone(), + product_sku: self + .product_sku + .clone() + .unwrap_or_else(|| CODEX_PRODUCT_SKU.to_string()), + http_clients: Arc::clone(http_clients), + } } } @@ -677,9 +694,12 @@ impl PluginsManager { // This assumes a single CODEX_HOME is only used by one product. let remote_installed_plugin_bundle_sync_gate = crate::remote::remote_installed_plugin_bundle_sync_gate(&codex_home); + let store = PluginStore::new(codex_home.clone()); + let tool_suggest_metadata_cache = + ToolSuggestMetadataCache::new(Arc::clone(&store.manifest_cache)); Self { - codex_home: codex_home.clone(), - store: PluginStore::new(codex_home), + codex_home, + store, featured_plugin_ids_cache: RwLock::new(None), recommended_plugins_cache: RwLock::new(HashMap::new()), recommended_plugins_refreshes: RwLock::new(HashMap::new()), @@ -695,7 +715,7 @@ impl PluginsManager { loaded_plugins_cache: Mutex::new(LoadedPluginsCache::default()), loaded_plugins_load_semaphore: Semaphore::new(/*permits*/ 1), skill_root_loader, - tool_suggest_metadata_cache: ToolSuggestMetadataCache::new(), + tool_suggest_metadata_cache, remote_installed_plugins_cache: RwLock::new(RemoteInstalledPluginsCache::default()), remote_installed_plugin_bundle_sync_gate, remote_installed_plugins_cache_refresh_state: RwLock::new( @@ -1147,6 +1167,7 @@ impl PluginsManager { plugin_id, &plugin_root, self.skill_root_loader.as_ref(), + &self.store.manifest_cache, ) .await } @@ -1168,6 +1189,7 @@ impl PluginsManager { plugin_id, &plugin_root, self.skill_root_loader.as_ref(), + &self.store.manifest_cache, ) .await } @@ -2425,7 +2447,11 @@ impl PluginsManager { && plugin.source.is_install_materialized() && let Some(plugin_id) = plugin_id.as_ref() && let Some(plugin_root) = self.store.active_plugin_root(plugin_id) - && let Some(manifest) = load_plugin_manifest(plugin_root.as_path()) + && let Some(manifest) = self + .store + .manifest_cache + .load(plugin_root.as_path()) + .map(|loaded| loaded.manifest) { local_version = manifest.version.clone(); let marketplace_category = interface @@ -2656,7 +2682,7 @@ impl PluginsManager { } let loaded_manifest = if codex_utils_plugins::find_plugin_manifest_path(source_path.as_path()).is_some() { - load_plugin_manifest_with_format(source_path.as_path()) + self.store.manifest_cache.load(source_path.as_path()) } else { plugin .manifest_fallback @@ -3600,7 +3626,8 @@ impl PluginsManager { config: &PluginsConfigInput, roots: &[AbsolutePathBuf], ) -> Result { - let mut outcome = list_marketplaces_with_home(roots, home_dir().as_deref())?; + let mut outcome = + list_marketplaces_with_cache(roots, home_dir().as_deref(), &self.store.manifest_cache)?; let policy = MarketplacePolicy::from_requirements(config.config_layer_stack.requirements()); outcome.marketplaces.retain(|marketplace| { policy @@ -3803,3 +3830,7 @@ impl PluginUninstallError { #[cfg(test)] #[path = "manager_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "plugins_config_input_tests.rs"] +mod plugins_config_input_tests; diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index d331e2939681..2ae390470d8d 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -8,6 +8,7 @@ use crate::ToolSuggestPluginDiscoveryInput; use crate::installed_marketplaces::marketplace_install_root; use crate::loader::load_plugin_skill_inventory; use crate::loader::load_plugins_from_layer_stack; +use crate::loader::plugin_capability_summary_from_root; use crate::loader::refresh_non_curated_plugin_cache; use crate::loader::refresh_non_curated_plugin_cache_force_reinstall; use crate::marketplace::MarketplacePluginInstallPolicy; @@ -1925,6 +1926,7 @@ async fn plugin_telemetry_metadata_uses_default_mcp_config_path() { &PluginId::parse("sample@test").expect("plugin id should parse"), &plugin_root.abs(), test_skill_root_loader().as_ref(), + &crate::manifest::ManifestCache::disabled(), ) .await; @@ -1968,6 +1970,7 @@ async fn plugin_capability_summary_uses_manifest_mcp_server_objects() { &PluginId::parse("counter-sample@test").expect("plugin id should parse"), &plugin_root.abs(), test_skill_root_loader().as_ref(), + &crate::manifest::ManifestCache::disabled(), ) .await; diff --git a/codex-rs/core-plugins/src/manifest.rs b/codex-rs/core-plugins/src/manifest.rs index 9991031f6056..a532660ddee7 100644 --- a/codex-rs/core-plugins/src/manifest.rs +++ b/codex-rs/core-plugins/src/manifest.rs @@ -10,6 +10,9 @@ use std::fs; use std::io; use std::path::Path; use std::path::PathBuf; + +mod manifest_cache; +pub(crate) use manifest_cache::ManifestCache; const MAX_DEFAULT_PROMPT_COUNT: usize = 3; const MAX_DEFAULT_PROMPT_LEN: usize = 128; @@ -165,41 +168,8 @@ pub fn is_agent_plugin_manifest(plugin_root: &Path) -> bool { } pub(crate) fn load_plugin_manifest_with_format(plugin_root: &Path) -> Option { - let manifest_path = find_plugin_manifest_path(plugin_root)?; - let contents = fs::read_to_string(&manifest_path).ok()?; - let is_agent_plugin = manifest_path == plugin_root.join(AGENT_PLUGIN_MANIFEST_RELATIVE_PATH); - let overlay = if is_agent_plugin { - let overlay_path = plugin_root.join(".codex-plugin/plugin.json"); - fs::read_to_string(&overlay_path) - .ok() - .map(|contents| (overlay_path, contents)) - } else { - None - }; - match parse_resolved_plugin_manifest( - plugin_root, - &manifest_path, - &contents, - overlay - .as_ref() - .map(|(path, contents)| (path.as_path(), contents.as_str())), - ) { - Ok(manifest) => Some(LoadedPluginManifest { - manifest, - format: if is_agent_plugin { - PluginManifestFormat::AgentPlugin - } else { - PluginManifestFormat::Legacy - }, - }), - Err(err) => { - tracing::warn!( - path = %manifest_path.display(), - "failed to parse plugin manifest: {err}" - ); - None - } - } + // TODO: accept an owner-provided cache when these standalone loaders join a repeated workflow. + ManifestCache::disabled().load(plugin_root) } pub(crate) fn load_plugin_command_paths(plugin_root: &Path) -> io::Result>> { @@ -220,15 +190,23 @@ pub(crate) fn load_plugin_command_paths(plugin_root: &Path) -> io::Result