diff --git a/internal/agent/lifecycle.go b/internal/agent/lifecycle.go index 7f408198..cc480c3f 100644 --- a/internal/agent/lifecycle.go +++ b/internal/agent/lifecycle.go @@ -4,10 +4,92 @@ import ( "context" "fmt" "strings" + "time" agentruntime "csgclaw/internal/runtime" ) +// restartCodexRuntime restarts the in-process Codex app-server without deleting +// its runtime directory. In particular, persisted Codex threads and conversation +// mappings survive profile changes such as reasoning effort updates. +func (s *Service) restartCodexRuntime(ctx context.Context, id string) (Agent, error) { + ctx, release, err := s.acquireAgentLifecycle(ctx, id) + if err != nil { + return Agent{}, err + } + defer release() + return s.restartCodexRuntimeLocked(ctx, id) +} + +func (s *Service) restartCodexRuntimeLocked(ctx context.Context, id string) (Agent, error) { + got, ok := s.Agent(id) + if !ok { + return Agent{}, fmt.Errorf("agent %q not found", strings.TrimSpace(id)) + } + if !strings.EqualFold(strings.TrimSpace(got.RuntimeKind), RuntimeKindCodex) { + return Agent{}, fmt.Errorf("agent %q is not a codex agent", got.ID) + } + + runtimeImpl, err := s.runtimeForKind(RuntimeKindCodex) + if err != nil { + return Agent{}, err + } + handle := runtimeHandleForAgent(got) + s.stopLifecycleAgent(got.ID) + if _, err := runtimeImpl.Stop(ctx, handle); err != nil { + return Agent{}, fmt.Errorf("stop codex agent for restart: %w", err) + } + if _, err := s.updateRuntimeState(got.ID, agentruntime.Info{ + HandleID: strings.TrimSpace(got.BoxID), + State: agentruntime.StateStopped, + }); err != nil { + return Agent{}, fmt.Errorf("save stopped codex agent state: %w", err) + } + if err := s.provisionRuntimeForAgent(ctx, runtimeImpl, got, ""); err != nil { + return Agent{}, fmt.Errorf("provision codex agent for restart: %w", err) + } + state, err := runtimeImpl.Start(ctx, handle) + if err != nil { + return Agent{}, fmt.Errorf("start codex agent after settings update: %w", err) + } + info, err := s.runtimeInfo(ctx, runtimeImpl, handle) + if err != nil { + return Agent{}, fmt.Errorf("read codex agent state after restart: %w", err) + } + if info.State == "" { + info.State = state + } + + s.mu.Lock() + current, key, ok := s.agentByIDLocked(id) + if !ok { + s.mu.Unlock() + return Agent{}, fmt.Errorf("agent %q not found", strings.TrimSpace(id)) + } + if handleID := strings.TrimSpace(info.HandleID); handleID != "" { + current.BoxID = handleID + } + current.Status = string(info.State) + current.AgentProfile.EnvRestartRequired = false + current.UpdatedAt = time.Now().UTC() + s.agents[key] = current + s.syncRuntimeRecordLocked(current) + err = s.saveLocked() + s.mu.Unlock() + if err != nil { + return Agent{}, err + } + + restarted, ok := s.Agent(id) + if !ok { + return Agent{}, fmt.Errorf("agent %q not found", strings.TrimSpace(id)) + } + if err := s.syncLifecycleForAgent(ctx, restarted); err != nil { + return Agent{}, err + } + return restarted, nil +} + type LifecycleObserver interface { EnsureAgent(context.Context, Agent) error StopAgent(string) diff --git a/internal/agent/service.go b/internal/agent/service.go index 95e9ec45..c0a9b24f 100644 --- a/internal/agent/service.go +++ b/internal/agent/service.go @@ -1678,6 +1678,9 @@ func (s *Service) Start(ctx context.Context, id string) (Agent, error) { if !ok { return Agent{}, fmt.Errorf("agent %q not found", id) } + if got.AgentProfile.EnvRestartRequired && !got.AgentProfile.ImageUpgradeRequired && strings.EqualFold(strings.TrimSpace(got.RuntimeKind), RuntimeKindCodex) { + return s.restartCodexRuntimeLocked(ctx, id) + } if got.AgentProfile.EnvRestartRequired || got.AgentProfile.ImageUpgradeRequired { return s.Recreate(ctx, id) } diff --git a/internal/agent/service_profiles.go b/internal/agent/service_profiles.go index 8b4f6b24..df1a17f0 100644 --- a/internal/agent/service_profiles.go +++ b/internal/agent/service_profiles.go @@ -135,7 +135,11 @@ func (s *Service) UpdateAgentProfile(id string, profile AgentProfile) (AgentProf return AgentProfileView{}, err } s.mu.Unlock() - if restartRequired && runtimeRunning && !isGatewayRuntimeKind(runtimeKind) { + if restartRequired && runtimeRunning && strings.EqualFold(runtimeKind, RuntimeKindCodex) { + if _, err := s.restartCodexRuntime(context.Background(), id); err != nil { + return AgentProfileView{}, err + } + } else if restartRequired && runtimeRunning && !isGatewayRuntimeKind(runtimeKind) { s.stopLifecycleAgent(id) } if err := s.syncGatewayAfterProfileChange(context.Background(), id, previous, normalized, restartRequired); err != nil { @@ -503,7 +507,11 @@ func (s *Service) update(ctx context.Context, id string, req UpdateRequest) (Age } } } - if restartRequired && runtimeRunning && !isGatewayRuntimeKind(runtimeKind) { + if restartRequired && runtimeRunning && strings.EqualFold(runtimeKind, RuntimeKindCodex) { + if _, err := s.restartCodexRuntime(ctx, id); err != nil { + return Agent{}, err + } + } else if restartRequired && runtimeRunning && !isGatewayRuntimeKind(runtimeKind) { s.stopLifecycleAgent(id) } @@ -511,7 +519,7 @@ func (s *Service) update(ctx context.Context, id string, req UpdateRequest) (Age if !ok { return Agent{}, fmt.Errorf("agent %q not found", id) } - if mcpServersUpdated && restartRequired && runtimeRunning && isManagerAgent(updated) { + if mcpServersUpdated && restartRequired && runtimeRunning && isManagerAgent(updated) && updated.AgentProfile.EnvRestartRequired { return s.Recreate(ctx, id) } if runtimeAffectingUpdate { diff --git a/internal/agent/service_test.go b/internal/agent/service_test.go index 4c628685..95e49087 100644 --- a/internal/agent/service_test.go +++ b/internal/agent/service_test.go @@ -1239,6 +1239,9 @@ func TestUpdateCodexAgentProfilePatchRestartsActiveBridge(t *testing.T) { if current.Profile.ModelID != "deepseek-v4-pro" { t.Fatalf("responses probe modelID = %q, want deepseek-v4-pro", current.Profile.ModelID) } + if current.Profile.ReasoningEffort != "high" { + t.Fatalf("responses probe reasoning effort = %q, want high", current.Profile.ReasoningEffort) + } return nil }, restart: func(change agentruntime.RuntimeConfigChange) (bool, error) { @@ -1246,6 +1249,10 @@ func TestUpdateCodexAgentProfilePatchRestartsActiveBridge(t *testing.T) { change.Previous.Profile.APIKey != change.Current.Profile.APIKey || change.Previous.Profile.ModelID != change.Current.Profile.ModelID, nil }, + del: func(context.Context, agentruntime.Handle) error { + t.Fatal("automatic Codex restart deleted the runtime and its history") + return nil + }, }), ) if err != nil { @@ -1273,10 +1280,11 @@ func TestUpdateCodexAgentProfilePatchRestartsActiveBridge(t *testing.T) { CreatedAt: time.Date(2026, 5, 18, 9, 0, 0, 0, time.UTC), } nextProfile := AgentProfile{ - Provider: ProviderAPI, - BaseURL: "https://api.deepseek.com", - APIKey: "deepseek-key", - ModelID: "deepseek-v4-pro", + Provider: ProviderAPI, + BaseURL: "https://api.deepseek.com", + APIKey: "deepseek-key", + ModelID: "deepseek-v4-pro", + ReasoningEffort: "high", } updated, err := svc.Update(context.Background(), "u-dev", UpdateRequest{AgentProfile: &nextProfile}) @@ -1286,12 +1294,15 @@ func TestUpdateCodexAgentProfilePatchRestartsActiveBridge(t *testing.T) { if probeCalls != 1 { t.Fatalf("responses probe calls = %d, want 1", probeCalls) } - if !updated.AgentProfile.EnvRestartRequired { - t.Fatal("Update().AgentProfile.EnvRestartRequired = false, want true so running Codex bridge is refreshed") + if updated.AgentProfile.EnvRestartRequired { + t.Fatal("Update().AgentProfile.EnvRestartRequired = true, want false after automatic Codex restart") } if len(observer.stopCalls) != 1 || observer.stopCalls[0] != "u-dev" { t.Fatalf("StopAgent() calls = %+v, want [u-dev]", observer.stopCalls) } + if len(observer.ensureCalls) != 1 || observer.ensureCalls[0].ID != "u-dev" { + t.Fatalf("EnsureAgent() calls = %+v, want one restart for u-dev", observer.ensureCalls) + } } func TestUpdateAgentProfileCodexRuntimeFallbackRestartsActiveBridge(t *testing.T) { @@ -1366,19 +1377,76 @@ func TestUpdateAgentProfileCodexRuntimeFallbackRestartsActiveBridge(t *testing.T if err != nil { t.Fatalf("UpdateAgentProfile() error = %v", err) } - if !view.EnvRestartRequired { - t.Fatal("UpdateAgentProfile().EnvRestartRequired = false, want true so running Codex bridge is refreshed") + if view.EnvRestartRequired { + t.Fatal("UpdateAgentProfile().EnvRestartRequired = true, want false after automatic Codex restart") } if len(observer.stopCalls) != 1 || observer.stopCalls[0] != "u-dev" { t.Fatalf("StopAgent() calls = %+v, want [u-dev]", observer.stopCalls) } - started, err := svc.Start(context.Background(), "u-dev") + if len(observer.ensureCalls) != 1 || observer.ensureCalls[0].ID != "u-dev" { + t.Fatalf("EnsureAgent() calls = %+v, want one automatic restart for u-dev", observer.ensureCalls) + } +} + +func TestCodexAutomaticRestartFailurePersistsStoppedAndRetriesWithoutDelete(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + startErr := errors.New("start failed") + deleteCalls := 0 + svc, err := NewService( + testModelConfig(), + config.ServerConfig{}, + "manager-image:test", + filepath.Join(t.TempDir(), "agents.json"), + WithRuntime(fakeAgentRuntime{ + kind: RuntimeKindCodex, + restart: func(agentruntime.RuntimeConfigChange) (bool, error) { return true, nil }, + start: func(context.Context, agentruntime.Handle) (agentruntime.State, error) { + if startErr != nil { + return agentruntime.StateStopped, startErr + } + return agentruntime.StateRunning, nil + }, + info: func(context.Context, agentruntime.Handle) (agentruntime.Info, error) { + return agentruntime.Info{State: agentruntime.StateRunning}, nil + }, + del: func(context.Context, agentruntime.Handle) error { + deleteCalls++ + return nil + }, + }), + ) + if err != nil { + t.Fatalf("NewService() error = %v", err) + } + profile := AgentProfile{Name: "dev", Provider: ProviderAPI, BaseURL: "https://old.example/v1", APIKey: "key", ModelID: "old", ProfileComplete: true} + svc.agents["u-dev"] = Agent{ID: "u-dev", Name: "dev", RuntimeID: "rt-u-dev", RuntimeKind: RuntimeKindCodex, Role: RoleWorker, Status: string(agentruntime.StateRunning), AgentProfile: profile, ProfileComplete: true} + + _, err = svc.Update(context.Background(), "u-dev", UpdateRequest{AgentProfile: &AgentProfile{Provider: ProviderAPI, BaseURL: "https://new.example/v1", APIKey: "key", ModelID: "new"}}) + if !errors.Is(err, startErr) { + t.Fatalf("Update() error = %v, want %v", err, startErr) + } + failed, ok := svc.agentSnapshot("u-dev") + if !ok { + t.Fatal("Agent() ok = false") + } + if got, want := failed.Status, string(agentruntime.StateStopped); got != want { + t.Fatalf("failed restart status = %q, want %q", got, want) + } + if !failed.AgentProfile.EnvRestartRequired { + t.Fatal("failed restart cleared EnvRestartRequired") + } + + startErr = nil + restarted, err := svc.Start(context.Background(), "u-dev") if err != nil { - t.Fatalf("Start() error = %v", err) + t.Fatalf("Start() retry error = %v", err) } - if started.AgentProfile.EnvRestartRequired { - t.Fatal("Start().AgentProfile.EnvRestartRequired = true, want false after recreate") + if restarted.AgentProfile.EnvRestartRequired || restarted.Status != string(agentruntime.StateRunning) { + t.Fatalf("Start() retry agent = %+v, want running without restart flag", restarted) + } + if deleteCalls != 0 { + t.Fatalf("Delete() calls = %d, want 0 to preserve history", deleteCalls) } } @@ -1742,12 +1810,15 @@ func TestUpdateCodexLocalWorkspaceDirMarksRunningRuntimeForRestart(t *testing.T) if err != nil { t.Fatalf("Update() error = %v", err) } - if !updated.AgentProfile.EnvRestartRequired { - t.Fatal("Update().AgentProfile.EnvRestartRequired = false, want true after codex local_workspace_dir change") + if updated.AgentProfile.EnvRestartRequired { + t.Fatal("Update().AgentProfile.EnvRestartRequired = true, want false after automatic Codex restart") } if len(observer.stopCalls) != 1 || observer.stopCalls[0] != "u-dev" { t.Fatalf("StopAgent() calls = %+v, want [u-dev]", observer.stopCalls) } + if len(observer.ensureCalls) != 1 || observer.ensureCalls[0].ID != "u-dev" { + t.Fatalf("EnsureAgent() calls = %+v, want one automatic restart for u-dev", observer.ensureCalls) + } } func TestAddMCPServersFromHubImportsRuntimeServersOnce(t *testing.T) { @@ -2139,7 +2210,7 @@ func TestAddMCPServersSerializesWithDirectMCPServersUpdate(t *testing.T) { } } -func TestAddMCPServersFromHubRecreatesRunningCodexManager(t *testing.T) { +func TestAddMCPServersFromHubRestartsRunningCodexManagerWithoutDeletingHistory(t *testing.T) { t.Setenv("HOME", t.TempDir()) origLocateCodexCLI := locateCodexCLI locateCodexCLI = func() (string, error) { return "/usr/local/bin/codex", nil } @@ -2204,15 +2275,15 @@ func TestAddMCPServersFromHubRecreatesRunningCodexManager(t *testing.T) { if err != nil { t.Fatalf("AddMCPServersFromHub() error = %v", err) } - if deleteCalls != 1 || newCalls != 1 { - t.Fatalf("manager recreate calls = delete %d/new %d, want 1/1", deleteCalls, newCalls) + if deleteCalls != 0 || newCalls != 0 { + t.Fatalf("manager restart calls = delete %d/new %d, want 0/0 to preserve history", deleteCalls, newCalls) } assertMCPServersHasServer(t, provisionedMCPServers, "context7") assertMCPServersHasServer(t, updated.MCPServers, "context7") if updated.AgentProfile.EnvRestartRequired { - t.Fatal("AddMCPServersFromHub().AgentProfile.EnvRestartRequired = true, want false after successful manager recreate") + t.Fatal("AddMCPServersFromHub().AgentProfile.EnvRestartRequired = true, want false after successful manager restart") } - if got, want := updated.BoxID, "codex-manager-session-new"; got != want { + if got, want := updated.BoxID, "codex-manager-session-old"; got != want { t.Fatalf("AddMCPServersFromHub().BoxID = %q, want %q", got, want) } } diff --git a/internal/api/handler_test.go b/internal/api/handler_test.go index b76de067..dea66f4e 100644 --- a/internal/api/handler_test.go +++ b/internal/api/handler_test.go @@ -12,9 +12,11 @@ import ( "net/http" "net/http/httptest" "os" + "os/exec" "path/filepath" "slices" "strings" + "sync" "testing" "time" @@ -205,11 +207,24 @@ func (apiFakeCodexBinaryProvider) Ensure(context.Context) (string, error) { return "/tmp/codex", nil } -type apiFakeCodexManager struct{} +type apiFakeCodexManager struct { + mu sync.Mutex + sessions map[string]*apiFakeCodexProcess +} + +type apiFakeCodexProcess struct { + cmd *exec.Cmd + session *codexruntime.Session +} -func (apiFakeCodexManager) Start(_ context.Context, spec codexruntime.SessionSpec) (*codexruntime.Session, error) { +func (m *apiFakeCodexManager) Start(_ context.Context, spec codexruntime.SessionSpec) (*codexruntime.Session, error) { + cmd := exec.Command(os.Args[0], "-test.run=^TestAPIFakeCodexProcess$") + cmd.Env = append(os.Environ(), "CSGCLAW_API_FAKE_CODEX_PROCESS=1") + if err := cmd.Start(); err != nil { + return nil, err + } now := time.Now().UTC() - return &codexruntime.Session{ + session := &codexruntime.Session{ RuntimeID: spec.RuntimeID, AgentID: spec.AgentID, AgentName: spec.AgentName, @@ -220,28 +235,70 @@ func (apiFakeCodexManager) Start(_ context.Context, spec codexruntime.SessionSpe HomeDir: spec.HomeDir, CodexHomeDir: spec.CodexHomeDir, StderrPath: spec.StderrPath, - ProcessID: os.Getpid(), + ProcessID: cmd.Process.Pid, CreatedAt: now, StartedAt: now, - }, nil + } + m.mu.Lock() + if m.sessions == nil { + m.sessions = make(map[string]*apiFakeCodexProcess) + } + m.sessions[spec.RuntimeID] = &apiFakeCodexProcess{cmd: cmd, session: session} + m.mu.Unlock() + return session, nil } -func (apiFakeCodexManager) Stop(context.Context, codexruntime.SessionHandle) error { +func (m *apiFakeCodexManager) Stop(_ context.Context, handle codexruntime.SessionHandle) error { + m.mu.Lock() + process := m.sessions[handle.RuntimeID] + delete(m.sessions, handle.RuntimeID) + m.mu.Unlock() + if process == nil || process.cmd == nil || process.cmd.Process == nil { + return os.ErrNotExist + } + _ = process.cmd.Process.Kill() + _ = process.cmd.Wait() return nil } -func (apiFakeCodexManager) LiveSession(codexruntime.SessionHandle) (*codexruntime.Session, error) { - return nil, os.ErrNotExist +func (m *apiFakeCodexManager) LiveSession(handle codexruntime.SessionHandle) (*codexruntime.Session, error) { + return m.Session(handle) } -func (apiFakeCodexManager) Session(codexruntime.SessionHandle) (*codexruntime.Session, error) { - return nil, os.ErrNotExist +func (m *apiFakeCodexManager) Session(handle codexruntime.SessionHandle) (*codexruntime.Session, error) { + m.mu.Lock() + defer m.mu.Unlock() + process := m.sessions[handle.RuntimeID] + if process == nil || process.session == nil { + return nil, os.ErrNotExist + } + cloned := *process.session + return &cloned, nil } -func (apiFakeCodexManager) Prompt(context.Context, codexruntime.SessionHandle, codexruntime.PromptRequest) (codexruntime.PromptResponse, error) { +func (*apiFakeCodexManager) Prompt(context.Context, codexruntime.SessionHandle, codexruntime.PromptRequest) (codexruntime.PromptResponse, error) { return codexruntime.PromptResponse{}, os.ErrNotExist } +func (m *apiFakeCodexManager) close() { + m.mu.Lock() + var handles []codexruntime.SessionHandle + for runtimeID := range m.sessions { + handles = append(handles, codexruntime.SessionHandle{RuntimeID: runtimeID}) + } + m.mu.Unlock() + for _, handle := range handles { + _ = m.Stop(context.Background(), handle) + } +} + +func TestAPIFakeCodexProcess(t *testing.T) { + if os.Getenv("CSGCLAW_API_FAKE_CODEX_PROCESS") != "1" { + return + } + select {} +} + type fakeCodexBridgeController struct { ensureCalls []agent.Agent refreshCalls []agentBindingRefreshCall @@ -2443,6 +2500,8 @@ func TestHandleAgentsMCPServersClosedLoopForSupportedRuntimes(t *testing.T) { codexRoot := t.TempDir() codexHomes := map[string]string{} statePath := filepath.Join(t.TempDir(), "agents.json") + codexManager := &apiFakeCodexManager{} + t.Cleanup(codexManager.close) codexRT := codexruntime.New(codexruntime.Dependencies{ BinaryProvider: apiFakeCodexBinaryProvider{}, AgentHome: func(agentID string) (string, error) { @@ -2458,7 +2517,7 @@ func TestHandleAgentsMCPServersClosedLoopForSupportedRuntimes(t *testing.T) { ModelID: "model-1", }) }, - Manager: apiFakeCodexManager{}, + Manager: codexManager, }) svc, err := agent.NewService( diff --git a/internal/runtime/codex/appserver_manager.go b/internal/runtime/codex/appserver_manager.go index f1cbc59a..37604323 100644 --- a/internal/runtime/codex/appserver_manager.go +++ b/internal/runtime/codex/appserver_manager.go @@ -126,6 +126,10 @@ func (m *appServerManager) Start(ctx context.Context, spec SessionSpec) (*Sessio logger := slog.New(slog.NewTextHandler(stderrFile, &slog.HandlerOptions{})) appClient := newAppServerClient(stdin, logger) + conversationSessions := cloneConversationSessions(spec.ConversationSessions) + if conversationSessions == nil { + conversationSessions = make(map[string]string) + } live := &liveSession{ cmd: cmd, stdin: stdin, @@ -133,7 +137,8 @@ func (m *appServerManager) Start(ctx context.Context, spec SessionSpec) (*Sessio done: make(chan struct{}), spec: spec, appClient: appClient, - conversationSessions: make(map[string]string), + conversationSessions: conversationSessions, + loadedConversations: make(map[string]bool), turnWaiters: make(map[string]*appServerTurnWaiter), turnThreads: make(map[string]string), commandOutputs: make(map[string]*appServerCommandOutputState), @@ -176,19 +181,20 @@ func (m *appServerManager) Start(ctx context.Context, spec SessionSpec) (*Sessio now := time.Now().UTC() session := &Session{ - RuntimeID: spec.RuntimeID, - AgentID: spec.AgentID, - AgentName: spec.AgentName, - SessionID: threadID, - BinaryPath: spec.BinaryPath, - RuntimeDir: spec.RuntimeDir, - WorkspaceDir: spec.WorkspaceDir, - HomeDir: spec.HomeDir, - CodexHomeDir: spec.CodexHomeDir, - StderrPath: spec.StderrPath, - ProcessID: cmd.Process.Pid, - CreatedAt: now, - StartedAt: now, + RuntimeID: spec.RuntimeID, + AgentID: spec.AgentID, + AgentName: spec.AgentName, + SessionID: threadID, + BinaryPath: spec.BinaryPath, + RuntimeDir: spec.RuntimeDir, + WorkspaceDir: spec.WorkspaceDir, + HomeDir: spec.HomeDir, + CodexHomeDir: spec.CodexHomeDir, + StderrPath: spec.StderrPath, + ProcessID: cmd.Process.Pid, + CreatedAt: now, + StartedAt: now, + ConversationSessions: cloneConversationSessions(spec.ConversationSessions), } live.mu.Lock() live.session = session @@ -365,23 +371,71 @@ func (m *appServerManager) EnsureSession(ctx context.Context, handle SessionHand } live.mu.Lock() - if threadID := strings.TrimSpace(live.conversationSessions[conversationKey]); threadID != "" { + if threadID := strings.TrimSpace(live.conversationSessions[conversationKey]); threadID != "" && live.loadedConversations[conversationKey] { live.mu.Unlock() return threadID, nil } + restoredThreadID := strings.TrimSpace(live.conversationSessions[conversationKey]) live.mu.Unlock() + if restoredThreadID != "" { + live.conversationResumeMu.Lock() + defer live.conversationResumeMu.Unlock() + live.conversationPersistMu.Lock() + defer live.conversationPersistMu.Unlock() + + live.mu.Lock() + if threadID := strings.TrimSpace(live.conversationSessions[conversationKey]); threadID != "" && live.loadedConversations[conversationKey] { + live.mu.Unlock() + return threadID, nil + } + restoredThreadID = strings.TrimSpace(live.conversationSessions[conversationKey]) + live.mu.Unlock() + + threadID, err := m.startOrResumeThread(ctx, live, restoredThreadID) + if err != nil { + return "", err + } + live.mu.Lock() + previous := live.conversationSessions[conversationKey] + live.conversationSessions[conversationKey] = threadID + live.loadedConversations[conversationKey] = true + conversations := cloneConversationSessions(live.conversationSessions) + live.mu.Unlock() + if err := m.persistConversationSessions(live, conversations); err != nil { + live.mu.Lock() + live.conversationSessions[conversationKey] = previous + delete(live.loadedConversations, conversationKey) + live.mu.Unlock() + return "", err + } + return threadID, nil + } threadID, err := m.startThread(ctx, live) if err != nil { return "", err } + live.conversationPersistMu.Lock() + defer live.conversationPersistMu.Unlock() live.mu.Lock() - defer live.mu.Unlock() if existing := strings.TrimSpace(live.conversationSessions[conversationKey]); existing != "" { + live.mu.Unlock() return existing, nil } live.conversationSessions[conversationKey] = threadID + live.loadedConversations[conversationKey] = true + conversations := cloneConversationSessions(live.conversationSessions) + live.mu.Unlock() + if err := m.persistConversationSessions(live, conversations); err != nil { + live.mu.Lock() + if live.conversationSessions[conversationKey] == threadID { + delete(live.conversationSessions, conversationKey) + delete(live.loadedConversations, conversationKey) + } + live.mu.Unlock() + return "", err + } return threadID, nil } @@ -400,10 +454,24 @@ func (m *appServerManager) ResetConversationHistory(ctx context.Context, handle return err } + live.conversationPersistMu.Lock() + defer live.conversationPersistMu.Unlock() live.mu.Lock() sessionID := strings.TrimSpace(live.conversationSessions[conversationKey]) delete(live.conversationSessions, conversationKey) + wasLoaded := live.loadedConversations[conversationKey] + delete(live.loadedConversations, conversationKey) + conversations := cloneConversationSessions(live.conversationSessions) live.mu.Unlock() + if err := m.persistConversationSessions(live, conversations); err != nil { + live.mu.Lock() + if sessionID != "" { + live.conversationSessions[conversationKey] = sessionID + live.loadedConversations[conversationKey] = wasLoaded + } + live.mu.Unlock() + return err + } if m.deps.Permission != nil && sessionID != "" { m.deps.Permission.CancelSession(runtimeID, sessionID) @@ -414,6 +482,13 @@ func (m *appServerManager) ResetConversationHistory(ctx context.Context, handle return nil } +func (m *appServerManager) persistConversationSessions(live *liveSession, conversations map[string]string) error { + if m == nil || live == nil || live.session == nil || m.deps.OnConversationSessionsChange == nil { + return nil + } + return m.deps.OnConversationSessionsChange(live.session, conversations) +} + func (m *appServerManager) ensureLiveSession(ctx context.Context, handle SessionHandle) (*liveSession, error) { runtimeID := strings.TrimSpace(handle.RuntimeID) if runtimeID == "" { diff --git a/internal/runtime/codex/appserver_manager_test.go b/internal/runtime/codex/appserver_manager_test.go index 3e639f08..c619a836 100644 --- a/internal/runtime/codex/appserver_manager_test.go +++ b/internal/runtime/codex/appserver_manager_test.go @@ -60,7 +60,13 @@ func TestAppServerManagerEnsureSessionCreatesConversationThread(t *testing.T) { withAppServerHelperCommand(t, "conversation-thread") dir := t.TempDir() spec := testAppServerSessionSpec(dir) - manager := newAppServerManager(testAppServerManagerDeps()) + var persisted []map[string]string + deps := testAppServerManagerDeps() + deps.OnConversationSessionsChange = func(_ *Session, conversations map[string]string) error { + persisted = append(persisted, cloneConversationSessions(conversations)) + return nil + } + manager := newAppServerManager(deps) session, err := manager.Start(context.Background(), spec) if err != nil { t.Fatalf("Start() error = %v", err) @@ -100,6 +106,91 @@ func TestAppServerManagerEnsureSessionCreatesConversationThread(t *testing.T) { if threadAfterReset == thread { t.Fatalf("conversation thread after reset = %q, want a new thread", threadAfterReset) } + if len(persisted) != 3 { + t.Fatalf("persisted conversation snapshots = %#v, want create/reset/recreate", persisted) + } + if got := persisted[0]["room-1"]; got != thread { + t.Fatalf("first persisted room thread = %q, want %q", got, thread) + } + if len(persisted[1]) != 0 { + t.Fatalf("reset persisted conversations = %#v, want empty", persisted[1]) + } + if got := persisted[2]["room-1"]; got != threadAfterReset { + t.Fatalf("recreated persisted room thread = %q, want %q", got, threadAfterReset) + } +} + +func TestAppServerManagerRestoresConversationThreadMapping(t *testing.T) { + withAppServerHelperCommand(t, "resume-success") + dir := t.TempDir() + spec := testAppServerSessionSpec(dir) + spec.ConversationSessions = map[string]string{"room-1": "persisted-room-thread"} + manager := newAppServerManager(testAppServerManagerDeps()) + if _, err := manager.Start(context.Background(), spec); err != nil { + t.Fatalf("Start() error = %v", err) + } + t.Cleanup(func() { _ = manager.Stop(context.Background(), SessionHandle{RuntimeID: spec.RuntimeID}) }) + + thread, err := manager.EnsureSession(context.Background(), SessionHandle{RuntimeID: spec.RuntimeID}, "room-1") + if err != nil { + t.Fatalf("EnsureSession() error = %v", err) + } + if got, want := thread, "resumed-thread"; got != want { + t.Fatalf("EnsureSession() = %q, want restored %q", got, want) + } + resp, err := manager.Prompt(context.Background(), SessionHandle{RuntimeID: spec.RuntimeID}, PromptRequest{ + SessionID: thread, + Prompt: []PromptContentBlock{TextBlock("continue")}, + }) + if err != nil { + t.Fatalf("Prompt(restored conversation) error = %v", err) + } + if resp.StopReason != StopReasonEndTurn { + t.Fatalf("Prompt(restored conversation) stop reason = %q, want %q", resp.StopReason, StopReasonEndTurn) + } +} + +func TestAppServerManagerSerializesConversationPersistence(t *testing.T) { + withAppServerHelperCommand(t, "conversation-thread") + dir := t.TempDir() + spec := testAppServerSessionSpec(dir) + entered := make(chan struct{}, 2) + release := make(chan struct{}) + deps := testAppServerManagerDeps() + deps.OnConversationSessionsChange = func(_ *Session, _ map[string]string) error { + entered <- struct{}{} + <-release + return nil + } + manager := newAppServerManager(deps) + if _, err := manager.Start(context.Background(), spec); err != nil { + t.Fatalf("Start() error = %v", err) + } + t.Cleanup(func() { _ = manager.Stop(context.Background(), SessionHandle{RuntimeID: spec.RuntimeID}) }) + + errCh := make(chan error, 2) + go func() { + _, err := manager.EnsureSession(context.Background(), SessionHandle{RuntimeID: spec.RuntimeID}, "room-1") + errCh <- err + }() + <-entered + go func() { + _, err := manager.EnsureSession(context.Background(), SessionHandle{RuntimeID: spec.RuntimeID}, "room-2") + errCh <- err + }() + select { + case <-entered: + t.Fatal("conversation persistence callbacks overlapped") + case <-time.After(50 * time.Millisecond): + } + release <- struct{}{} + <-entered + release <- struct{}{} + for range 2 { + if err := <-errCh; err != nil { + t.Fatalf("EnsureSession() error = %v", err) + } + } } func TestAppServerManagerEnsureSessionHandlesThreadNotificationBeforeResponse(t *testing.T) { diff --git a/internal/runtime/codex/config_controller.go b/internal/runtime/codex/config_controller.go index 3792d494..c925bc1c 100644 --- a/internal/runtime/codex/config_controller.go +++ b/internal/runtime/codex/config_controller.go @@ -7,6 +7,7 @@ import ( "net/http" "os" "path/filepath" + "reflect" "regexp" "strings" "sync" @@ -59,7 +60,8 @@ func (r *Runtime) ValidateConfig(ctx context.Context, current agentruntime.Runti } func (r *Runtime) RestartRequired(change agentruntime.RuntimeConfigChange) (bool, error) { - return codexWorkspaceOptionChanged(change.Previous.Options, change.Current.Options), nil + return codexWorkspaceOptionChanged(change.Previous.Options, change.Current.Options) || + !reflect.DeepEqual(change.Previous.Profile, change.Current.Profile), nil } func (r *Runtime) ReconcileConfig(ctx context.Context, h agentruntime.Handle, change agentruntime.RuntimeConfigChange) error { diff --git a/internal/runtime/codex/runtime.go b/internal/runtime/codex/runtime.go index 60e19741..f6639596 100644 --- a/internal/runtime/codex/runtime.go +++ b/internal/runtime/codex/runtime.go @@ -59,16 +59,17 @@ type AgentRef struct { } type SessionSpec struct { - RuntimeID string - AgentID string - AgentName string - BinaryPath string - RuntimeDir string - WorkspaceDir string - HomeDir string - CodexHomeDir string - StderrPath string - Profile agentruntime.Profile + RuntimeID string + AgentID string + AgentName string + BinaryPath string + RuntimeDir string + WorkspaceDir string + HomeDir string + CodexHomeDir string + StderrPath string + Profile agentruntime.Profile + ConversationSessions map[string]string } type SessionHandle struct { @@ -76,20 +77,21 @@ type SessionHandle struct { } type Session struct { - RuntimeID string - AgentID string - AgentName string - SessionID string - BinaryPath string - RuntimeDir string - WorkspaceDir string - HomeDir string - CodexHomeDir string - StderrPath string - ProcessID int - CreatedAt time.Time - StartedAt time.Time - AgentCapabilities any + RuntimeID string + AgentID string + AgentName string + SessionID string + BinaryPath string + RuntimeDir string + WorkspaceDir string + HomeDir string + CodexHomeDir string + StderrPath string + ProcessID int + CreatedAt time.Time + StartedAt time.Time + AgentCapabilities any + ConversationSessions map[string]string } type Manager interface { @@ -355,11 +357,18 @@ func (r *Runtime) Start(ctx context.Context, h agentruntime.Handle) (agentruntim return agentruntime.StateUnknown, err } agentRef.Profile = agentRef.Profile.Normalized() + var conversationSessions map[string]string + if sessionMeta, readErr := r.readSessionMetadata(strings.TrimSpace(h.RuntimeID)); readErr == nil { + conversationSessions = cloneConversationSessions(sessionMeta.ConversationSessions) + } else if !errors.Is(readErr, os.ErrNotExist) { + return agentruntime.StateUnknown, fmt.Errorf("read persisted codex conversations: %w", readErr) + } session, err := r.ensureSession(ctx, SessionSpec{ - RuntimeID: strings.TrimSpace(h.RuntimeID), - AgentID: strings.TrimSpace(agentRef.ID), - AgentName: strings.TrimSpace(agentRef.Name), - Profile: agentRef.Profile, + RuntimeID: strings.TrimSpace(h.RuntimeID), + AgentID: strings.TrimSpace(agentRef.ID), + AgentName: strings.TrimSpace(agentRef.Name), + Profile: agentRef.Profile, + ConversationSessions: conversationSessions, }) if err != nil { if sessionRestoreErr != nil { @@ -530,6 +539,14 @@ func (r *Runtime) sessionManager() Manager { } _ = writeJSONFile(r.writeFile, filepath.Join(session.RuntimeDir, runtimeFileName), meta) }, + OnConversationSessionsChange: func(session *Session, conversations map[string]string) error { + if session == nil { + return nil + } + meta := sessionToSessionMetadata(session) + meta.ConversationSessions = cloneConversationSessions(conversations) + return writeJSONFile(r.writeFile, filepath.Join(session.RuntimeDir, sessionFileName), meta) + }, }) manager.deps.HydrateSession = func(ctx context.Context, handle SessionHandle) (*Session, error) { return r.hydratePersistedSession(ctx, manager, handle) @@ -643,7 +660,8 @@ func (r *Runtime) hydratePersistedSession(ctx context.Context, manager *appServe if err != nil { return nil, err } - if _, err := r.readSessionMetadata(runtimeID); err != nil { + sessionMeta, err := r.readSessionMetadata(runtimeID) + if err != nil { return nil, err } agentRef, err := r.resolveAgent(agentruntime.Handle{RuntimeID: runtimeID}) @@ -669,16 +687,17 @@ func (r *Runtime) hydratePersistedSession(ctx context.Context, manager *appServe return nil, fmt.Errorf("resolve codex binary: %w", err) } spec := SessionSpec{ - RuntimeID: runtimeID, - AgentID: agentID, - AgentName: firstNonEmpty(agentRef.Name, meta.AgentName), - BinaryPath: binaryPath, - RuntimeDir: dirs.Root, - WorkspaceDir: workspaceDir, - HomeDir: r.hostSessionHomeDir(dirs.Home), - CodexHomeDir: dirs.CodexHome, - StderrPath: dirs.StderrLog, - Profile: agentRef.Profile.Normalized(), + RuntimeID: runtimeID, + AgentID: agentID, + AgentName: firstNonEmpty(agentRef.Name, meta.AgentName), + BinaryPath: binaryPath, + RuntimeDir: dirs.Root, + WorkspaceDir: workspaceDir, + HomeDir: r.hostSessionHomeDir(dirs.Home), + CodexHomeDir: dirs.CodexHome, + StderrPath: dirs.StderrLog, + Profile: agentRef.Profile.Normalized(), + ConversationSessions: cloneConversationSessions(sessionMeta.ConversationSessions), } if err := r.mkdirAll(spec.WorkspaceDir, 0o755); err != nil { return nil, fmt.Errorf("create codex workspace dir %s: %w", spec.WorkspaceDir, err) @@ -1699,12 +1718,13 @@ type runtimeMetadata struct { } type sessionMetadata struct { - RuntimeID string `json:"runtime_id"` - SessionID string `json:"session_id"` - WorkspaceDir string `json:"workspace_dir"` - HomeDir string `json:"home_dir"` - CodexHomeDir string `json:"codex_home_dir"` - StartedAt time.Time `json:"started_at,omitempty"` + RuntimeID string `json:"runtime_id"` + SessionID string `json:"session_id"` + WorkspaceDir string `json:"workspace_dir"` + HomeDir string `json:"home_dir"` + CodexHomeDir string `json:"codex_home_dir"` + StartedAt time.Time `json:"started_at,omitempty"` + ConversationSessions map[string]string `json:"conversation_sessions,omitempty"` } func sessionToRuntimeMetadata(session *Session) runtimeMetadata { @@ -1723,15 +1743,34 @@ func sessionToRuntimeMetadata(session *Session) runtimeMetadata { func sessionToSessionMetadata(session *Session) sessionMetadata { return normalizeSessionMetadata(sessionMetadata{ - RuntimeID: session.RuntimeID, - SessionID: session.SessionID, - WorkspaceDir: session.WorkspaceDir, - HomeDir: session.HomeDir, - CodexHomeDir: session.CodexHomeDir, - StartedAt: session.StartedAt, + RuntimeID: session.RuntimeID, + SessionID: session.SessionID, + WorkspaceDir: session.WorkspaceDir, + HomeDir: session.HomeDir, + CodexHomeDir: session.CodexHomeDir, + StartedAt: session.StartedAt, + ConversationSessions: cloneConversationSessions(session.ConversationSessions), }) } +func cloneConversationSessions(in map[string]string) map[string]string { + if len(in) == 0 { + return nil + } + out := make(map[string]string, len(in)) + for key, value := range in { + key = strings.TrimSpace(key) + value = strings.TrimSpace(value) + if key != "" && value != "" { + out[key] = value + } + } + if len(out) == 0 { + return nil + } + return out +} + func normalizeRuntimeMetadata(meta runtimeMetadata) runtimeMetadata { meta.RuntimeID = strings.TrimSpace(meta.RuntimeID) meta.AgentID = strings.TrimSpace(meta.AgentID) @@ -1757,6 +1796,7 @@ func normalizeSessionMetadata(meta sessionMetadata) sessionMetadata { meta.WorkspaceDir = strings.TrimSpace(meta.WorkspaceDir) meta.HomeDir = strings.TrimSpace(meta.HomeDir) meta.CodexHomeDir = strings.TrimSpace(meta.CodexHomeDir) + meta.ConversationSessions = cloneConversationSessions(meta.ConversationSessions) if !meta.StartedAt.IsZero() { meta.StartedAt = meta.StartedAt.UTC() } @@ -1833,12 +1873,13 @@ func writeJSONFile(writeFile func(string, []byte, os.FileMode) error, path strin } type managerDeps struct { - EventSink SessionEventSink - Permission PermissionBroker - UserInput UserInputBroker - OpenFile func(string, int, os.FileMode) (*os.File, error) - WriteFile func(string, []byte, os.FileMode) error - ReadFile func(string) ([]byte, error) - OnExit func(*Session, int) - HydrateSession func(context.Context, SessionHandle) (*Session, error) + EventSink SessionEventSink + Permission PermissionBroker + UserInput UserInputBroker + OpenFile func(string, int, os.FileMode) (*os.File, error) + WriteFile func(string, []byte, os.FileMode) error + ReadFile func(string) ([]byte, error) + OnExit func(*Session, int) + HydrateSession func(context.Context, SessionHandle) (*Session, error) + OnConversationSessionsChange func(*Session, map[string]string) error } diff --git a/internal/runtime/codex/runtime_test.go b/internal/runtime/codex/runtime_test.go index d2b37b9d..2b8cb209 100644 --- a/internal/runtime/codex/runtime_test.go +++ b/internal/runtime/codex/runtime_test.go @@ -333,25 +333,38 @@ func TestRestartRequiredReturnsTrueWhenLocalWorkspaceDirChanges(t *testing.T) { } } -func TestRestartRequiredIgnoresProfileChanges(t *testing.T) { - rt := &Runtime{} - got, err := rt.RestartRequired(agentruntime.RuntimeConfigChange{ - Previous: agentruntime.RuntimeConfigSnapshot{ - Profile: agentruntime.RuntimeProfileConfig{ - ModelID: "gpt-5.5", - }, +func TestRestartRequiredReturnsTrueWhenProfileChanges(t *testing.T) { + tests := []struct { + name string + previous agentruntime.RuntimeProfileConfig + current agentruntime.RuntimeProfileConfig + }{ + { + name: "model", + previous: agentruntime.RuntimeProfileConfig{ModelID: "qwen3.6-plus"}, + current: agentruntime.RuntimeProfileConfig{ModelID: "claude-sonnet-4-7"}, }, - Current: agentruntime.RuntimeConfigSnapshot{ - Profile: agentruntime.RuntimeProfileConfig{ - ModelID: "gpt-5.6", - }, + { + name: "reasoning effort", + previous: agentruntime.RuntimeProfileConfig{ReasoningEffort: "medium"}, + current: agentruntime.RuntimeProfileConfig{ReasoningEffort: "high"}, }, - }) - if err != nil { - t.Fatalf("RestartRequired() error = %v", err) } - if got { - t.Fatal("RestartRequired() = true, want false when only profile changes") + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + rt := &Runtime{} + got, err := rt.RestartRequired(agentruntime.RuntimeConfigChange{ + Previous: agentruntime.RuntimeConfigSnapshot{Profile: test.previous}, + Current: agentruntime.RuntimeConfigSnapshot{Profile: test.current}, + }) + if err != nil { + t.Fatalf("RestartRequired() error = %v", err) + } + if !got { + t.Fatal("RestartRequired() = false, want true after Codex profile change") + } + }) } } @@ -3108,6 +3121,68 @@ func TestRuntimeStartKeepsExistingRunningSession(t *testing.T) { } } +func TestRuntimeStartRestoresPersistedConversationMappings(t *testing.T) { + root := t.TempDir() + var startedSpec SessionSpec + startCalls := 0 + rt := New(Dependencies{ + BinaryProvider: fakeBinaryProvider{path: "/tmp/codex"}, + AgentHome: func(agentID string) (string, error) { + return filepath.Join(root, agentID), nil + }, + ResolveAgent: func(h agentruntime.Handle) (AgentRef, error) { + return AgentRef{ID: "u-alice", Name: "alice", RuntimeID: h.RuntimeID}, nil + }, + Manager: fakeManager{start: func(_ context.Context, spec SessionSpec) (*Session, error) { + startCalls++ + startedSpec = spec + return &Session{ + RuntimeID: spec.RuntimeID, + AgentID: spec.AgentID, + AgentName: spec.AgentName, + SessionID: "main-thread", + RuntimeDir: spec.RuntimeDir, + WorkspaceDir: spec.WorkspaceDir, + HomeDir: spec.HomeDir, + CodexHomeDir: spec.CodexHomeDir, + StderrPath: spec.StderrPath, + ConversationSessions: cloneConversationSessions(spec.ConversationSessions), + CreatedAt: time.Now().UTC(), + StartedAt: time.Now().UTC(), + }, nil + }}, + }) + handle, err := rt.New(context.Background(), agentruntime.Spec{ + RuntimeID: "rt-u-alice", + AgentID: "u-alice", + AgentName: "alice", + }) + if err != nil { + t.Fatalf("New() error = %v", err) + } + runtimeDir := filepath.Join(root, "agent-alice", ".codex") + if err := writeJSONFile(os.WriteFile, filepath.Join(runtimeDir, sessionFileName), sessionMetadata{ + RuntimeID: "rt-u-alice", + SessionID: "main-thread", + ConversationSessions: map[string]string{"room-1": "room-thread"}, + }); err != nil { + t.Fatalf("write session metadata: %v", err) + } + + if _, err := rt.Stop(context.Background(), handle); err != nil { + t.Fatalf("Stop() error = %v", err) + } + if _, err := rt.Start(context.Background(), handle); err != nil { + t.Fatalf("Start() error = %v", err) + } + if startCalls != 2 { + t.Fatalf("manager Start() calls = %d, want 2", startCalls) + } + if got, want := startedSpec.ConversationSessions["room-1"], "room-thread"; got != want { + t.Fatalf("Start() conversation mapping = %q, want %q", got, want) + } +} + func TestRuntimeStartRepairsFailedPersistedSession(t *testing.T) { root := t.TempDir() oldProcess := exec.Command("sleep", "30") diff --git a/internal/runtime/codex/session_manager.go b/internal/runtime/codex/session_manager.go index 300e798c..d918d394 100644 --- a/internal/runtime/codex/session_manager.go +++ b/internal/runtime/codex/session_manager.go @@ -10,6 +10,8 @@ import ( type liveSession struct { mu sync.Mutex + conversationResumeMu sync.Mutex + conversationPersistMu sync.Mutex session *Session appClient *appServerClient cmd *exec.Cmd @@ -18,6 +20,7 @@ type liveSession struct { done chan struct{} spec SessionSpec conversationSessions map[string]string + loadedConversations map[string]bool turnWaiters map[string]*appServerTurnWaiter turnThreads map[string]string turnThreadOrder []string