From bebaee35125120adc7cb55ae9aa59d31fb6bd26a Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Sun, 25 Oct 2026 09:04:42 +0000 Subject: [PATCH] Keep the Codex state database on compute-local disk Codex opens its SQLite databases on every app-server start. Under the hosted state directory, which lives on NFS workspace storage, a new Codex Session's native_prepare took 1.48-1.74 s on mx (n=5); with CODEX_SQLITE_HOME under the Runtime home on the VM's own disk it took 0.38-0.44 s (n=5). Nothing the Session needs is lost when the compute is replaced: Codex rebuilds the thread index and history views from the rollouts in CODEX_HOME and reads dynamic tools from the rollout, and goals, whose state lives only in SQLite, are disabled. --- apps/daemon/internal/agent/codex/declaration.go | 2 +- .../internal/agent/codex/error_classification_test.go | 4 ++-- apps/daemon/internal/agent/codex/executor.go | 8 ++++---- apps/daemon/internal/agent/codex/mcp_config_test.go | 4 ++-- .../internal/agent/codex/mcp_http_bearer_test.go | 4 ++-- .../internal/agent/codex/mcp_http_preflight_test.go | 2 +- apps/daemon/internal/agent/codex/mcp_http_test.go | 10 +++++----- apps/daemon/internal/agent/codex/options.go | 8 ++++---- .../internal/agent/codex/permission_profile_test.go | 10 ++++++---- .../internal/agent/codex/preparation_helpers_test.go | 2 +- .../internal/agent/codex/provider_config_test.go | 6 +++--- apps/daemon/internal/agent/codex/recovery_test.go | 2 +- apps/daemon/internal/agent/codex/rpc_close_test.go | 2 +- apps/daemon/internal/agent/codex/session.go | 4 +++- .../internal/agent/codex/session_cancel_test.go | 2 +- .../internal/agent/codex/session_cancel_write_test.go | 2 +- .../agent/codex/session_command_output_test.go | 2 +- .../daemon/internal/agent/codex/session_items_test.go | 2 +- .../internal/agent/codex/session_messages_test.go | 2 +- .../agent/codex/session_notifications_rpc_test.go | 4 ++-- .../agent/codex/session_notifications_test.go | 6 +++--- apps/daemon/internal/agent/codex/session_plan.go | 11 ++++++++++- .../internal/agent/codex/session_policy_test.go | 2 +- .../daemon/internal/agent/codex/session_tools_test.go | 2 +- .../internal/agent/codex/session_usage_live_test.go | 6 +++--- .../daemon/internal/agent/codex/session_usage_test.go | 8 ++++---- .../agent/codex/subagent_observations_test.go | 2 +- .../daemon/internal/dispatch/functions_native_test.go | 2 +- services/core/deploy/codex/README.md | 2 +- 29 files changed, 68 insertions(+), 55 deletions(-) diff --git a/apps/daemon/internal/agent/codex/declaration.go b/apps/daemon/internal/agent/codex/declaration.go index ff96549d3..c580a1f24 100644 --- a/apps/daemon/internal/agent/codex/declaration.go +++ b/apps/daemon/internal/agent/codex/declaration.go @@ -33,7 +33,7 @@ func discoverWithCheck(parent context.Context, options agent.DiscoveryOptions, i for _, capability := range []*proto.CapabilitySupport{&caps.LocalEnvironment, &caps.WorkspaceReadPreparation, &caps.WorkspaceOutputExport, &caps.RetainedNativeHistory} { agent.Narrow(capability, local) } - runtime.Executor = NewExecutorFactory(options.StateRoot) + runtime.Executor = NewExecutorFactory(options.StateRoot, options.RuntimeRoot) fmt.Fprintf(options.Stdout, "Codex preflight ok (%s)\n", version) return runtime } diff --git a/apps/daemon/internal/agent/codex/error_classification_test.go b/apps/daemon/internal/agent/codex/error_classification_test.go index bceba9b8a..055317606 100644 --- a/apps/daemon/internal/agent/codex/error_classification_test.go +++ b/apps/daemon/internal/agent/codex/error_classification_test.go @@ -50,7 +50,7 @@ func TestClassificationOnlyFromRootTerminalError(t *testing.T) { for _, mode := range []string{"failed", "recovered", "notification-only"} { t.Run(mode, func(t *testing.T) { out := make(chan proto.Envelope, 16) - s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t)), rpc: NewJSONRPCClient(JSONRPCConfig{})} + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), rpc: NewJSONRPCClient(JSONRPCConfig{})} s.registerHandlers() s.setThreadID("root") scopeNotification(t, s, "turn/started", `{"threadId":"root","turn":{"id":"turn"}}`) @@ -109,7 +109,7 @@ func TestProviderRejectionClassifiedOnce(t *testing.T) { } } out := make(chan proto.Envelope, 16) - s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t)), rpc: NewJSONRPCClient(JSONRPCConfig{})} + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), rpc: NewJSONRPCClient(JSONRPCConfig{})} s.registerHandlers() s.setThreadID("root") turnError, _ := json.Marshal(TurnError{Message: body, CodexErrorInfo: json.RawMessage(`"other"`)}) diff --git a/apps/daemon/internal/agent/codex/executor.go b/apps/daemon/internal/agent/codex/executor.go index 43f687af9..3bd077a98 100644 --- a/apps/daemon/internal/agent/codex/executor.go +++ b/apps/daemon/internal/agent/codex/executor.go @@ -21,8 +21,8 @@ type Executor struct { closeMu sync.Mutex } -func PrepareExecutor(ctx context.Context, stateRoot string, req proto.PromptRequestPayload) (agent.Executor, error) { - return newExecutor(ctx, req, defaultSessionConfig(stateRoot)) +func PrepareExecutor(ctx context.Context, stateRoot, runtimeRoot string, req proto.PromptRequestPayload) (agent.Executor, error) { + return newExecutor(ctx, req, defaultSessionConfig(stateRoot, runtimeRoot)) } func newExecutor(ctx context.Context, req proto.PromptRequestPayload, cfg sessionConfig) (*Executor, error) { @@ -172,8 +172,8 @@ func (e *Executor) Close(ctx context.Context) error { var _ agent.Executor = (*Executor)(nil) -func NewExecutorFactory(stateRoot string) agent.ExecutorFactory { +func NewExecutorFactory(stateRoot, runtimeRoot string) agent.ExecutorFactory { return func(ctx context.Context, req proto.PromptRequestPayload) (agent.Executor, error) { - return PrepareExecutor(ctx, stateRoot, req) + return PrepareExecutor(ctx, stateRoot, runtimeRoot, req) } } diff --git a/apps/daemon/internal/agent/codex/mcp_config_test.go b/apps/daemon/internal/agent/codex/mcp_config_test.go index c55ca481a..d05689bb0 100644 --- a/apps/daemon/internal/agent/codex/mcp_config_test.go +++ b/apps/daemon/internal/agent/codex/mcp_config_test.go @@ -73,11 +73,11 @@ func TestWriteCodexMCPConfig_EmitsStreamableHTTPURL(t *testing.T) { // TestWriteCodexMCPConfig_FreshHomeDropsStaleEntries documents the // "fresh entries only" guarantee: callers allocate a brand-new -// CODEX_HOME per prompt (BuildSessionPlan does this via allocCodexHome +// CODEX_HOME per prompt (BuildSessionPlan does this via allocSessionDir // + plan.Cleanup), so the previous run's mcp_servers can't leak. // // writeCodexMCPConfig itself is APPEND semantics now — the truncation -// guarantee lives in allocCodexHome's RemoveAll, not in the writer. +// guarantee lives in allocSessionDir's RemoveAll, not in the writer. // Test that workflow explicitly so a future refactor that breaks the // fresh-home contract fails here. func TestWriteCodexMCPConfig_FreshHomeDropsStaleEntries(t *testing.T) { diff --git a/apps/daemon/internal/agent/codex/mcp_http_bearer_test.go b/apps/daemon/internal/agent/codex/mcp_http_bearer_test.go index c7dcaf0b5..da0627df2 100644 --- a/apps/daemon/internal/agent/codex/mcp_http_bearer_test.go +++ b/apps/daemon/internal/agent/codex/mcp_http_bearer_test.go @@ -22,7 +22,7 @@ func TestMCPHTTPBearerPlanSeparatesServersAndProcesses(t *testing.T) { req := proto.PromptRequestPayload{AgentStateKey: "retained-mcp", DisableExecutionEnvironment: true, MCPHTTPServers: &servers} seen := map[string]bool{} for range 2 { - plan, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t))) + plan, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t), t.TempDir())) if err != nil { t.Fatal(err) } @@ -58,7 +58,7 @@ func TestMCPHTTPBearerRejectsInvalidTokensWithoutPersistence(t *testing.T) { for _, token := range []string{"", "=", " has-space", "has-space ", "has space", "line\r\ninjection", "nul\x00byte", "opaque中文", "middle=padding", "punctuation:invalid"} { servers := []proto.MCPHTTPServer{{ConnectionOrigin: "service", ServerLabel: "tools", ServerURL: "https://tools.example/mcp", BearerToken: &token}} req := proto.PromptRequestPayload{AgentStateKey: "invalid-bearer", DisableExecutionEnvironment: true, MCPHTTPServers: &servers} - if _, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t))); err == nil || err.Error() != "invalid HTTPS MCP bearer credential" { + if _, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t), t.TempDir())); err == nil || err.Error() != "invalid HTTPS MCP bearer credential" { t.Fatal("invalid bearer value accepted or unsafe error returned") } } diff --git a/apps/daemon/internal/agent/codex/mcp_http_preflight_test.go b/apps/daemon/internal/agent/codex/mcp_http_preflight_test.go index 297725749..7af9982d5 100644 --- a/apps/daemon/internal/agent/codex/mcp_http_preflight_test.go +++ b/apps/daemon/internal/agent/codex/mcp_http_preflight_test.go @@ -174,7 +174,7 @@ func TestPublicMCPHTTPPreparationChecksBeforeNewAndResumedThread(t *testing.T) { t.Fatal(err) } assertPreparationOnly(t, root) - home, err := allocCodexHome(testStateRoot(t), req.AgentStateKey) + home, err := allocSessionDir(testStateRoot(t), req.AgentStateKey) if err != nil { t.Fatal(err) } diff --git a/apps/daemon/internal/agent/codex/mcp_http_test.go b/apps/daemon/internal/agent/codex/mcp_http_test.go index 75610d9fa..81d00996f 100644 --- a/apps/daemon/internal/agent/codex/mcp_http_test.go +++ b/apps/daemon/internal/agent/codex/mcp_http_test.go @@ -21,7 +21,7 @@ func TestPublicMCPHTTPPlanOwnsConfigurationAndPreservesHistory(t *testing.T) { {ConnectionOrigin: "service", ServerLabel: "blocked", ServerURL: "http://127.0.0.1:12345/mcp", AllowedTools: &denyAll}, } req := proto.PromptRequestPayload{AgentStateKey: "public-mcp", DisableExecutionEnvironment: true, MCPHTTPServers: &servers} - plan, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t))) + plan, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t), t.TempDir())) if err != nil { t.Fatal(err) } @@ -31,7 +31,7 @@ func TestPublicMCPHTTPPlanOwnsConfigurationAndPreservesHistory(t *testing.T) { !slices.Contains(plan.ExtraConfig, [2]string{"shell_environment_policy.exclude", `["OAC_RUNTIME_MCP_*"]`}) { t.Fatal("native profile kept only credential isolation") } - home, err := allocCodexHome(testStateRoot(t), req.AgentStateKey) + home, err := allocSessionDir(testStateRoot(t), req.AgentStateKey) if err != nil { t.Fatal(err) } @@ -56,7 +56,7 @@ func TestPublicMCPHTTPPlanOwnsConfigurationAndPreservesHistory(t *testing.T) { t.Fatal(err) } servers = []proto.MCPHTTPServer{{ConnectionOrigin: "service", ServerLabel: "replacement", ServerURL: "https://new.example/mcp"}} - second, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t))) + second, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t), t.TempDir())) if err != nil { t.Fatal(err) } @@ -91,7 +91,7 @@ func TestPublicMCPHTTPRejectsInvalidProfileAndStoredCredentials(t *testing.T) { } } t.Setenv("OAC_RUNTIME_HOME", t.TempDir()) - home, err := allocCodexHome(testStateRoot(t), "credentials") + home, err := allocSessionDir(testStateRoot(t), "credentials") if err != nil { t.Fatal(err) } @@ -101,7 +101,7 @@ func TestPublicMCPHTTPRejectsInvalidProfileAndStoredCredentials(t *testing.T) { t.Fatal(err) } req := proto.PromptRequestPayload{AgentStateKey: "credentials", DisableExecutionEnvironment: true, MCPHTTPServers: &valid} - if _, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t))); err == nil { + if _, _, err := prepareSessionPlan(t.Context(), req, defaultSessionConfig(testStateRoot(t), t.TempDir())); err == nil { t.Fatal("existing MCP credentials accepted") } if after, err := os.ReadFile(path); err != nil || !reflect.DeepEqual(after, stored) { diff --git a/apps/daemon/internal/agent/codex/options.go b/apps/daemon/internal/agent/codex/options.go index 81da2df19..00331a6db 100644 --- a/apps/daemon/internal/agent/codex/options.go +++ b/apps/daemon/internal/agent/codex/options.go @@ -101,7 +101,7 @@ func BuildSessionPlan(stateRoot, agentStateKey string, opts map[string]any, cont plan.Model = stringOpt(opts, "model") plan.SystemPrompt = stringOpt(opts, "system_prompt") - codexHome, err := allocCodexHome(stateRoot, agentStateKey) + codexHome, err := allocSessionDir(stateRoot, agentStateKey) if err != nil { return plan, err } @@ -212,9 +212,9 @@ func applyExecutionControls(plan *SessionPlan, controls *proto.ExecutionControls // helpers // --------------------------------------------------------------------------- -func allocCodexHome(root, agentStateKey string) (string, error) { +func allocSessionDir(root, agentStateKey string) (string, error) { if strings.TrimSpace(agentStateKey) == "" { - return "", fmt.Errorf("codex: agentStateKey required for CODEX_HOME allocation") + return "", fmt.Errorf("codex: agentStateKey required for a Session directory") } if !filepath.IsAbs(root) || filepath.Clean(root) != root { return "", fmt.Errorf("codex: state root must be a clean absolute directory") @@ -232,7 +232,7 @@ func allocCodexHome(root, agentStateKey string) (string, error) { dirParts := append([]string{root, "daemon", "agent-sessions"}, safeParts...) dir := filepath.Join(dirParts...) if err := os.MkdirAll(dir, 0o700); err != nil { - return "", fmt.Errorf("codex: create CODEX_HOME %s: %w", dir, err) + return "", fmt.Errorf("codex: create Session directory %s: %w", dir, err) } return dir, nil } diff --git a/apps/daemon/internal/agent/codex/permission_profile_test.go b/apps/daemon/internal/agent/codex/permission_profile_test.go index 722bda5bd..cf9794976 100644 --- a/apps/daemon/internal/agent/codex/permission_profile_test.go +++ b/apps/daemon/internal/agent/codex/permission_profile_test.go @@ -12,7 +12,7 @@ func TestRuntimeUsesHostPermissions(t *testing.T) { t.Setenv("OAC_RUNTIME_HOME", t.TempDir()) { req := proto.PromptRequestPayload{AgentStateKey: "session", DisableSubagents: true, LocalEnvironment: &proto.LocalEnvironment{WorkspaceRoot: t.TempDir()}} - plan, _, err := prepareSessionPlan(t.Context(), req, sessionConfig{stateRoot: testStateRoot(t)}) + plan, _, err := prepareSessionPlan(t.Context(), req, sessionConfig{stateRoot: testStateRoot(t), runtimeRoot: t.TempDir()}) if err != nil { t.Fatal(err) } @@ -32,12 +32,13 @@ func TestSelfHostedToolEnvironmentCannotRedirectNativeHistory(t *testing.T) { root := t.TempDir() t.Setenv("OAC_RUNTIME_HOME", root) config := filepath.Join(root, "tool-env.json") - if err := os.WriteFile(config, []byte(`{"CODEX_HOME":"wrong","HOME":"wrong","USERPROFILE":"wrong","USER_VALUE":"ready"}`), 0600); err != nil { + if err := os.WriteFile(config, []byte(`{"CODEX_HOME":"wrong","CODEX_SQLITE_HOME":"wrong","HOME":"wrong","USERPROFILE":"wrong","USER_VALUE":"ready"}`), 0600); err != nil { t.Fatal(err) } t.Setenv("OAC_RUNTIME_TOOL_ENV_FILE", config) req := proto.PromptRequestPayload{AgentStateKey: "session", DisableSubagents: true, LocalEnvironment: &proto.LocalEnvironment{}} - plan, _, err := prepareSessionPlan(t.Context(), req, sessionConfig{stateRoot: testStateRoot(t)}) + runtimeRoot := t.TempDir() + plan, _, err := prepareSessionPlan(t.Context(), req, sessionConfig{stateRoot: testStateRoot(t), runtimeRoot: runtimeRoot}) if err != nil { t.Fatal(err) } @@ -47,7 +48,8 @@ func TestSelfHostedToolEnvironmentCannotRedirectNativeHistory(t *testing.T) { key, value, _ := strings.Cut(entry, "=") values[key] = value } - if values["CODEX_HOME"] == "" || values["CODEX_HOME"] == "wrong" || values["HOME"] == "wrong" || values["USERPROFILE"] == "wrong" || values["USER_VALUE"] != "ready" { + if values["CODEX_HOME"] == "" || values["CODEX_HOME"] == "wrong" || values["HOME"] == "wrong" || values["USERPROFILE"] == "wrong" || values["USER_VALUE"] != "ready" || + values["CODEX_SQLITE_HOME"] != filepath.Join(runtimeRoot, "daemon", "agent-sessions", "session") { t.Fatal("initialization changed native state identity") } } diff --git a/apps/daemon/internal/agent/codex/preparation_helpers_test.go b/apps/daemon/internal/agent/codex/preparation_helpers_test.go index 780febc73..d61484e14 100644 --- a/apps/daemon/internal/agent/codex/preparation_helpers_test.go +++ b/apps/daemon/internal/agent/codex/preparation_helpers_test.go @@ -37,7 +37,7 @@ func preparationFixture(t *testing.T) (proto.PromptRequestPayload, sessionConfig if err := os.WriteFile(binary, []byte(body), 0o700); err != nil { t.Fatal(err) } - cfg := defaultSessionConfig(testStateRoot(t)) + cfg := defaultSessionConfig(testStateRoot(t), t.TempDir()) cfg.codexBinary = binary req := proto.PromptRequestPayload{ AgentKind: "codex", AgentStateKey: "prepared-session", diff --git a/apps/daemon/internal/agent/codex/provider_config_test.go b/apps/daemon/internal/agent/codex/provider_config_test.go index 8a1f76213..a0c7e033a 100644 --- a/apps/daemon/internal/agent/codex/provider_config_test.go +++ b/apps/daemon/internal/agent/codex/provider_config_test.go @@ -263,7 +263,7 @@ func TestProviderCredentialOnlyInNativeLoopEnvironment(t *testing.T) { func TestExplicitStateRootSurvivesRuntimeReplacement(t *testing.T) { state := t.TempDir() t.Setenv("OAC_RUNTIME_HOME", t.TempDir()) - first, err := allocCodexHome(state, "stable-session") + first, err := allocSessionDir(state, "stable-session") if err != nil { t.Fatal(err) } @@ -272,7 +272,7 @@ func TestExplicitStateRootSurvivesRuntimeReplacement(t *testing.T) { t.Fatal(err) } t.Setenv("OAC_RUNTIME_HOME", t.TempDir()) - second, err := allocCodexHome(state, "stable-session") + second, err := allocSessionDir(state, "stable-session") if err != nil || second != first { t.Fatal("instance identity changed native history", err) } @@ -283,7 +283,7 @@ func TestExplicitStateRootSurvivesRuntimeReplacement(t *testing.T) { if err = os.WriteFile(unavailable, nil, 0600); err != nil { t.Fatal(err) } - if _, err = allocCodexHome(unavailable, "stable-session"); err == nil { + if _, err = allocSessionDir(unavailable, "stable-session"); err == nil { t.Fatal("unavailable explicit root fell back") } } diff --git a/apps/daemon/internal/agent/codex/recovery_test.go b/apps/daemon/internal/agent/codex/recovery_test.go index 82ced82cf..5c6431a15 100644 --- a/apps/daemon/internal/agent/codex/recovery_test.go +++ b/apps/daemon/internal/agent/codex/recovery_test.go @@ -147,7 +147,7 @@ func TestPreparedRecoveryCannotStartWithoutExistingHistory(t *testing.T) { t.Run(environment, func(t *testing.T) { req, cfg, root := preparationFixture(t) req.RequireExistingNativeSession = true - cwd, err := allocCodexHome(testStateRoot(t), req.AgentStateKey) + cwd, err := allocSessionDir(testStateRoot(t), req.AgentStateKey) if err != nil { t.Fatal(err) } diff --git a/apps/daemon/internal/agent/codex/rpc_close_test.go b/apps/daemon/internal/agent/codex/rpc_close_test.go index 7d4c2c29b..6eb6c841e 100644 --- a/apps/daemon/internal/agent/codex/rpc_close_test.go +++ b/apps/daemon/internal/agent/codex/rpc_close_test.go @@ -80,7 +80,7 @@ func TestJSONRPCClientCloseCanRetryUnreapedChild(t *testing.T) { closeConcurrently(true) ownerCtx, cancelOwner := context.WithCancel(t.Context()) s := &Session{rpc: client, cancelCtx: ownerCtx, cancelFn: cancelOwner, - cfg: defaultSessionConfig(testStateRoot(t)), bufs: NewItemBuffers()} + cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), bufs: NewItemBuffers()} ctx, cancelWait := context.WithTimeout(t.Context(), 20*time.Millisecond) defer cancelWait() if err := s.Cancel(ctx); !errors.Is(err, context.DeadlineExceeded) { diff --git a/apps/daemon/internal/agent/codex/session.go b/apps/daemon/internal/agent/codex/session.go index e844581f0..d5832b64a 100644 --- a/apps/daemon/internal/agent/codex/session.go +++ b/apps/daemon/internal/agent/codex/session.go @@ -24,14 +24,16 @@ const terminalSendTimeout = 2 * time.Second // defaultSessionConfig. type sessionConfig struct { stateRoot string + runtimeRoot string codexBinary string logger *slog.Logger killTimeout time.Duration } -func defaultSessionConfig(stateRoot string) sessionConfig { +func defaultSessionConfig(stateRoot, runtimeRoot string) sessionConfig { return sessionConfig{ stateRoot: stateRoot, + runtimeRoot: runtimeRoot, codexBinary: defaultBinary(), logger: obslog.Bg(), killTimeout: rpcKillTimeout, diff --git a/apps/daemon/internal/agent/codex/session_cancel_test.go b/apps/daemon/internal/agent/codex/session_cancel_test.go index e4307ae1e..5bcc706e4 100644 --- a/apps/daemon/internal/agent/codex/session_cancel_test.go +++ b/apps/daemon/internal/agent/codex/session_cancel_test.go @@ -116,7 +116,7 @@ func cancellationTestSession(t *testing.T) (*Session, *TestClient, ServerSide) { ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) s := &Session{rpc: client.JSONRPCClient, cancelCtx: ctx, cancelFn: cancel, - cfg: defaultSessionConfig(testStateRoot(t)), out: make(chan proto.Envelope, 8), bufs: NewItemBuffers()} + cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), out: make(chan proto.Envelope, 8), bufs: NewItemBuffers()} return s, client, server } diff --git a/apps/daemon/internal/agent/codex/session_cancel_write_test.go b/apps/daemon/internal/agent/codex/session_cancel_write_test.go index e9adb4970..11c547aca 100644 --- a/apps/daemon/internal/agent/codex/session_cancel_write_test.go +++ b/apps/daemon/internal/agent/codex/session_cancel_write_test.go @@ -146,7 +146,7 @@ func TestCancelReleasesBlockedNativeProcess(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() s := &Session{rpc: client, cancelCtx: ctx, cancelFn: cancel, - cfg: defaultSessionConfig(testStateRoot(t)), bufs: NewItemBuffers()} + cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), bufs: NewItemBuffers()} s.setThreadID("native-thread") s.onTurnStarted(json.RawMessage(`{"threadId":"native-thread","turn":{"id":"native-turn"}}`)) cancelDone := make(chan struct{}) diff --git a/apps/daemon/internal/agent/codex/session_command_output_test.go b/apps/daemon/internal/agent/codex/session_command_output_test.go index f14b3eab3..8f3c42421 100644 --- a/apps/daemon/internal/agent/codex/session_command_output_test.go +++ b/apps/daemon/internal/agent/codex/session_command_output_test.go @@ -10,7 +10,7 @@ import ( func TestCommandOutputRequiresRootTurn(t *testing.T) { out := make(chan proto.Envelope, 10) - s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t)), rpc: NewJSONRPCClient(JSONRPCConfig{})} + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), rpc: NewJSONRPCClient(JSONRPCConfig{})} s.registerHandlers() s.setThreadID("root") s.beginRootTurn("root", "turn") diff --git a/apps/daemon/internal/agent/codex/session_items_test.go b/apps/daemon/internal/agent/codex/session_items_test.go index 95ce60d56..0c29fff53 100644 --- a/apps/daemon/internal/agent/codex/session_items_test.go +++ b/apps/daemon/internal/agent/codex/session_items_test.go @@ -71,7 +71,7 @@ func TestDispatchStartedItem_UnknownSilent(t *testing.T) { func TestReasoningIsReportedOnlyWhenRequested(t *testing.T) { for _, requested := range []bool{true, false} { out := make(chan proto.Envelope, 16) - s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t)), + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), rpc: NewJSONRPCClient(JSONRPCConfig{}), bufs: NewItemBuffers(), reasoningSummary: requested} s.registerHandlers() s.setThreadID("root") diff --git a/apps/daemon/internal/agent/codex/session_messages_test.go b/apps/daemon/internal/agent/codex/session_messages_test.go index eacc37199..2b908a5d3 100644 --- a/apps/daemon/internal/agent/codex/session_messages_test.go +++ b/apps/daemon/internal/agent/codex/session_messages_test.go @@ -10,7 +10,7 @@ import ( func TestMessageObservationKeepsNativeBoundaries(t *testing.T) { out := make(chan proto.Envelope, 16) - s := &Session{runID: "run", out: out, cancelCtx: context.Background(), bufs: NewItemBuffers(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), bufs: NewItemBuffers(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("thread") s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`)) s.onItemStarted(json.RawMessage(`{"threadId":"thread","turnId":"turn","item":{"type":"agentMessage","id":"a","phase":"commentary"}}`)) diff --git a/apps/daemon/internal/agent/codex/session_notifications_rpc_test.go b/apps/daemon/internal/agent/codex/session_notifications_rpc_test.go index c14d7cd98..bc6b4da2a 100644 --- a/apps/daemon/internal/agent/codex/session_notifications_rpc_test.go +++ b/apps/daemon/internal/agent/codex/session_notifications_rpc_test.go @@ -19,7 +19,7 @@ func TestRootRPCNotificationOrdering(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() out := make(chan proto.Envelope, 16) - s := &Session{runID: "run", out: out, rpc: client.JSONRPCClient, cancelCtx: ctx, bufs: NewItemBuffers(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{runID: "run", out: out, rpc: client.JSONRPCClient, cancelCtx: ctx, bufs: NewItemBuffers(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.registerHandlers() barrier := make(chan struct{}, 1) client.OnNotification("test/barrier", func(json.RawMessage) { barrier <- struct{}{} }) @@ -132,7 +132,7 @@ func TestThreadRPCResponseRequiresRootIdentity(t *testing.T) { defer cleanup() ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - s := &Session{rpc: client.JSONRPCClient, cancelCtx: ctx, cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{rpc: client.JSONRPCClient, cancelCtx: ctx, cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.registerHandlers() result := make(chan error, 1) go func() { result <- s.resumeThread("root", SessionPlan{}) }() diff --git a/apps/daemon/internal/agent/codex/session_notifications_test.go b/apps/daemon/internal/agent/codex/session_notifications_test.go index 3e33d7885..e8bf661b2 100644 --- a/apps/daemon/internal/agent/codex/session_notifications_test.go +++ b/apps/daemon/internal/agent/codex/session_notifications_test.go @@ -12,7 +12,7 @@ func TestRootNotificationIsolation(t *testing.T) { for _, childStarted := range []bool{false, true} { t.Run(map[bool]string{false: "child without thread started", true: "child thread started"}[childStarted], func(t *testing.T) { out := make(chan proto.Envelope, 64) - s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t)), + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), rpc: NewJSONRPCClient(JSONRPCConfig{}), bufs: NewItemBuffers(), reasoningSummary: true} s.registerHandlers() s.setThreadID("root") @@ -96,7 +96,7 @@ func TestRootNotificationIsolation(t *testing.T) { } func TestNotificationsCannotEstablishRootIdentity(t *testing.T) { - s := &Session{cfg: defaultSessionConfig(testStateRoot(t)), rpc: NewJSONRPCClient(JSONRPCConfig{}), bufs: NewItemBuffers()} + s := &Session{cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), rpc: NewJSONRPCClient(JSONRPCConfig{}), bufs: NewItemBuffers()} s.registerHandlers() scopeNotification(t, s, "thread/started", `{"thread":{"id":"unrelated"}}`) scopeNotification(t, s, "turn/started", `{"threadId":"unrelated","turn":{"id":"unrelated-turn"}}`) @@ -116,7 +116,7 @@ func TestRootNativeErrorNotification(t *testing.T) { for _, location := range []string{"error notification", "completion"} { t.Run(name+"/"+location, func(t *testing.T) { out := make(chan proto.Envelope, 4) - s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t)), rpc: NewJSONRPCClient(JSONRPCConfig{})} + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), rpc: NewJSONRPCClient(JSONRPCConfig{})} s.registerHandlers() s.setThreadID("root") scopeNotification(t, s, "turn/started", `{"threadId":"root","turn":{"id":"turn"}}`) diff --git a/apps/daemon/internal/agent/codex/session_plan.go b/apps/daemon/internal/agent/codex/session_plan.go index fe240ec20..e7d055c40 100644 --- a/apps/daemon/internal/agent/codex/session_plan.go +++ b/apps/daemon/internal/agent/codex/session_plan.go @@ -24,6 +24,15 @@ func prepareSessionPlan(ctx context.Context, req proto.PromptRequestPayload, cfg if err != nil { return SessionPlan{}, nil, fmt.Errorf("codex: build session plan: %w", err) } + // Codex's SQLite databases stay on the compute's own disk, where they open faster than + // on shared state storage. A replacement compute rebuilds them from the history in + // CODEX_HOME; goals, whose state lives only there, are disabled. + sqliteHome, err := allocSessionDir(cfg.runtimeRoot, req.AgentStateKey) + if err != nil { + plan.Cleanup() + return SessionPlan{}, nil, err + } + plan.Env = append(plan.Env, "CODEX_SQLITE_HOME="+sqliteHome) if err := configureSubagentObservations(&plan, req); err != nil { plan.Cleanup() return SessionPlan{}, nil, err @@ -47,7 +56,7 @@ func prepareSessionPlan(ctx context.Context, req proto.PromptRequestPayload, cfg return SessionPlan{}, nil, err } for key, value := range values { - if strings.EqualFold(key, "CODEX_HOME") || strings.EqualFold(key, "HOME") || strings.EqualFold(key, "USERPROFILE") { + if strings.EqualFold(key, "CODEX_HOME") || strings.EqualFold(key, "CODEX_SQLITE_HOME") || strings.EqualFold(key, "HOME") || strings.EqualFold(key, "USERPROFILE") { continue } plan.Env = append(plan.Env, key+"="+value) diff --git a/apps/daemon/internal/agent/codex/session_policy_test.go b/apps/daemon/internal/agent/codex/session_policy_test.go index f534f4cf4..697dd9df2 100644 --- a/apps/daemon/internal/agent/codex/session_policy_test.go +++ b/apps/daemon/internal/agent/codex/session_policy_test.go @@ -16,7 +16,7 @@ func TestThreadRequestsApplyDeploymentPolicy(t *testing.T) { for _, method := range []string{"thread/start", "thread/resume"} { t.Run(method, func(t *testing.T) { t.Setenv("OAC_RUNTIME_HOME", t.TempDir()) - plan, _, err := prepareSessionPlan(context.Background(), proto.PromptRequestPayload{AgentStateKey: "conv/agent/codex", DisableSubagents: true, DisableExecutionEnvironment: true}, sessionConfig{stateRoot: testStateRoot(t)}) + plan, _, err := prepareSessionPlan(context.Background(), proto.PromptRequestPayload{AgentStateKey: "conv/agent/codex", DisableSubagents: true, DisableExecutionEnvironment: true}, sessionConfig{stateRoot: testStateRoot(t), runtimeRoot: t.TempDir()}) if err != nil { t.Fatal(err) } diff --git a/apps/daemon/internal/agent/codex/session_tools_test.go b/apps/daemon/internal/agent/codex/session_tools_test.go index 460a5fe4f..96622746f 100644 --- a/apps/daemon/internal/agent/codex/session_tools_test.go +++ b/apps/daemon/internal/agent/codex/session_tools_test.go @@ -29,7 +29,7 @@ func TestToolObservations(t *testing.T) { } t.Run(source.ID, func(t *testing.T) { out := make(chan proto.Envelope, 4) - s := &Session{runID: "run", out: out, cancelCtx: context.Background(), bufs: NewItemBuffers(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), bufs: NewItemBuffers(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("private-thread") s.onTurnStarted(json.RawMessage(`{"threadId":"private-thread","turn":{"id":"private-turn"}}`)) raw := json.RawMessage(`{"threadId":"private-thread","turnId":"private-turn","item":` + item + `}`) diff --git a/apps/daemon/internal/agent/codex/session_usage_live_test.go b/apps/daemon/internal/agent/codex/session_usage_live_test.go index ba9fe754f..55ab89a61 100644 --- a/apps/daemon/internal/agent/codex/session_usage_live_test.go +++ b/apps/daemon/internal/agent/codex/session_usage_live_test.go @@ -15,7 +15,7 @@ const activeUsageNotification = `{"method":"thread/tokenUsage/updated","params": func TestUsagePublishedBeforeCompletion(t *testing.T) { out := make(chan proto.Envelope, 8) - s := &Session{runID: "run", out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{runID: "run", out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("thread") s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"old","tokenUsage":{"total":{"inputTokens":10,"cachedInputTokens":1,"outputTokens":3,"reasoningOutputTokens":1,"totalTokens":13}}}`)) if len(s.out) != 0 { @@ -159,7 +159,7 @@ func TestLiveUsageRequiresConsistentCompleteBreakdown(t *testing.T) { } { t.Run(name, func(t *testing.T) { out := make(chan proto.Envelope, 1) - s := &Session{out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("thread") s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`)) s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"turn","tokenUsage":{"total":` + counters + `}}`)) @@ -177,7 +177,7 @@ func TestLiveUsageRequiresConsistentCompleteBreakdown(t *testing.T) { func TestLiveUsageRejectsInconsistentResumeDelta(t *testing.T) { out := make(chan proto.Envelope, 2) - s := &Session{out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("thread") s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"old","tokenUsage":{"total":{"inputTokens":100,"cachedInputTokens":90,"outputTokens":50,"reasoningOutputTokens":10,"totalTokens":150}}}`)) s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`)) diff --git a/apps/daemon/internal/agent/codex/session_usage_test.go b/apps/daemon/internal/agent/codex/session_usage_test.go index 7c5e17ada..473a02ab1 100644 --- a/apps/daemon/internal/agent/codex/session_usage_test.go +++ b/apps/daemon/internal/agent/codex/session_usage_test.go @@ -62,7 +62,7 @@ func TestNativeTokenUsageAcrossRuns(t *testing.T) { } func TestNativeTokenUsageFreshThreadAndIgnoredPayloads(t *testing.T) { - s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("thread") s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"current"}}`)) s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"current","tokenUsage":{"total":{"inputTokens":321,"outputTokens":45}}}`)) @@ -81,7 +81,7 @@ func TestNativeTokenUsageFreshThreadAndIgnoredPayloads(t *testing.T) { } func TestLegacyTurnUsagePayload(t *testing.T) { - s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("thread") s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"current"}}`)) s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","usage":{"inputTokens":123,"outputTokens":12}}`)) @@ -95,7 +95,7 @@ func TestLegacyTurnUsagePayload(t *testing.T) { } func TestCompleteTokenBreakdownAndCancellation(t *testing.T) { - s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("thread") s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"old","tokenUsage":{"total":{"inputTokens":100,"cachedInputTokens":20,"outputTokens":50,"reasoningOutputTokens":10,"totalTokens":150}}}`)) s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"new"}}`)) @@ -164,7 +164,7 @@ func TestUnadvancedThreadTotalIsNotTurnUsage(t *testing.T) { } { t.Run(name, func(t *testing.T) { out := make(chan proto.Envelope, 8) - s := &Session{runID: "run", out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t))} + s := &Session{runID: "run", out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig(testStateRoot(t), t.TempDir())} s.setThreadID("thread") total := `{"inputTokens":0,"cachedInputTokens":0,"outputTokens":0,"reasoningOutputTokens":0,"totalTokens":0}` if replay != "" { diff --git a/apps/daemon/internal/agent/codex/subagent_observations_test.go b/apps/daemon/internal/agent/codex/subagent_observations_test.go index 14589a919..0175d8f29 100644 --- a/apps/daemon/internal/agent/codex/subagent_observations_test.go +++ b/apps/daemon/internal/agent/codex/subagent_observations_test.go @@ -74,7 +74,7 @@ func observationSession(t *testing.T, status string) (*Session, *subagentFixture t.Fatal(err) } f.persist(t) - s := &Session{runID: "run", nativeHome: f.home, rpc: client.JSONRPCClient, out: out, cancelCtx: ctx, cancelFn: cancel, cfg: defaultSessionConfig(testStateRoot(t)), bufs: NewItemBuffers()} + s := &Session{runID: "run", nativeHome: f.home, rpc: client.JSONRPCClient, out: out, cancelCtx: ctx, cancelFn: cancel, cfg: defaultSessionConfig(testStateRoot(t), t.TempDir()), bufs: NewItemBuffers()} s.setThreadID("root") s.beginRootTurn("root", "root-turn") s.startSubagentObservations() diff --git a/apps/daemon/internal/dispatch/functions_native_test.go b/apps/daemon/internal/dispatch/functions_native_test.go index b022d94dd..056576d93 100644 --- a/apps/daemon/internal/dispatch/functions_native_test.go +++ b/apps/daemon/internal/dispatch/functions_native_test.go @@ -108,7 +108,7 @@ func TestNativeFunctionBridge(t *testing.T) { })) defer model.Close() reg := agent.NewRegistry() - registerExecutorKind(reg, proto.SupportedAgentKind{Kind: "codex", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{FunctionTools: proto.CapabilitySupported, EnvironmentNone: proto.CapabilitySupported})}, codex.NewExecutorFactory(t.TempDir())) + registerExecutorKind(reg, proto.SupportedAgentKind{Kind: "codex", Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{FunctionTools: proto.CapabilitySupported, EnvironmentNone: proto.CapabilitySupported})}, codex.NewExecutorFactory(t.TempDir(), t.TempDir())) sender := make(nativeFunctionSender, 256) ctx, cancel := context.WithTimeout(t.Context(), 60*time.Second) defer cancel() diff --git a/services/core/deploy/codex/README.md b/services/core/deploy/codex/README.md index 0637c19bc..ac2763cbf 100644 --- a/services/core/deploy/codex/README.md +++ b/services/core/deploy/codex/README.md @@ -12,7 +12,7 @@ The adapter accepts only Codex `0.153.4`: installation and recovery checks requi Codex accepts only the `responses` model protocol ([`harnessconfig/codex`](../../../../internal/harnessconfig/codex/declaration.go)). A Chat Completions or Anthropic provider is rejected; nothing converts between protocols. [Model execution](../../../../contracts/agents-api/model-execution.md#deployment-defaults) owns provider selection, including the per-Harness deployment default. -Each Session has its own `CODEX_HOME` at `/daemon/agent-sessions//`, where the Runtime supplies `` from the [state directory setting](../../../../docs/configuration.md#runtime-resource-directories). Native history stays there; Runtime device credentials stay in the Runtime home. The state layout alone does not qualify cross-node native continuation. The adapter regenerates that directory's `config.toml` on every prompt: the Session's frozen provider bundle becomes the `[model_providers.oac]` block, with `wire_api` `responses`, and the thread is pinned to that provider, so Codex never falls back to its built-in `openai` provider. Provider bearer credentials use the native `env_key` reference and enter the model-loop child process environment; they are not written into retained `config.toml`. The adapter excludes that variable from native shell environments with `shell_environment_policy.exclude` and disables `shell_snapshot`, whose capture in the pinned native version can otherwise persist the model-loop environment. This retains native thread history without retaining native shell snapshots; it introduces no user setting. +Each Session has its own `CODEX_HOME` at `/daemon/agent-sessions//`, where the Runtime supplies `` from the [state directory setting](../../../../docs/configuration.md#runtime-resource-directories). Native history stays there; Runtime device credentials stay in the Runtime home. Codex's SQLite databases, `CODEX_SQLITE_HOME`, take the same path under the Runtime home on the compute's own disk, because on NFS workspace storage they made preparing each new Session about 1.1 s slower. Compute replacement loses nothing the Session needs: Codex rebuilds its thread index and history views from the rollouts in `CODEX_HOME` and reads a thread's dynamic tools from its rollout, goals are disabled as described below, and the adapter uses none of the other databases. The state layout alone does not qualify cross-node native continuation. The adapter regenerates that directory's `config.toml` on every prompt: the Session's frozen provider bundle becomes the `[model_providers.oac]` block, with `wire_api` `responses`, and the thread is pinned to that provider, so Codex never falls back to its built-in `openai` provider. Provider bearer credentials use the native `env_key` reference and enter the model-loop child process environment; they are not written into retained `config.toml`. The adapter excludes that variable from native shell environments with `shell_environment_policy.exclude` and disables `shell_snapshot`, whose capture in the pinned native version can otherwise persist the model-loop environment. This retains native thread history without retaining native shell snapshots; it introduces no user setting. By default Codex retries a failed model connection forever, waiting up to 60 seconds between attempts, so a Turn whose model host is unreachable stays in progress with no events. The adapter disables that `unbounded_connection_retries` feature, so Codex falls back to its bounded defaults: up to 4 retries per request and 5 per response stream. A model host that does not resolve or refuses connections then fails the Turn with `connection_failed` within about a minute, depending on how long each lookup or connection attempt takes. Retries start only after a request fails, so slow responses are unaffected, and a stream still ends after Codex's 300 seconds without data. Codex sets no connect timeout: against a host that silently drops connections, each of the 30 attempts waits for the operating system's connect timeout.