Merge pull request #631 from Yumiue/codex/fix-sqlite-session-concurrency

fix(session): 串行化 SQLite 会话写入

Author: phantom5099

internal/cli/gateway_runtime_bridge.go

@@ -44,6 +44,26 @@ type runtimeSessionCreator interface {
 	CreateSession(ctx context.Context, id string) (agentsession.Session, error)
 }
 
+type runtimeSessionDeleter interface {
+	SupportsSessionMutationBoundary() bool
+	DeleteSession(ctx context.Context, sessionID string) error
+}
+
+type runtimeSessionRenamer interface {
+	SupportsSessionMutationBoundary() bool
+	RenameSession(ctx context.Context, sessionID string, title string) error
+}
+
+type runtimeSessionModelUpdater interface {
+	SupportsSessionMutationBoundary() bool
+	UpdateSessionModel(ctx context.Context, sessionID string, providerID string, modelID string) error
+}
+
+type runtimeSessionsProviderModelSyncer interface {
+	SupportsSessionMutationBoundary() bool
+	SyncSessionsProviderModel(ctx context.Context, providerID string, modelID string) error
+}
+
 type runtimeTodoLister interface {
 	ListTodos(ctx context.Context, sessionID string) (agentruntime.TodoSnapshot, error)
 }
@@ -648,6 +668,12 @@ func (b *gatewayRuntimePortBridge) DeleteSession(ctx context.Context, input gate
 	if sessionID == "" {
 		return false, gateway.ErrRuntimeResourceNotFound
 	}
+	if deleter, ok := b.runtime.(runtimeSessionDeleter); ok {
+		if err := deleter.DeleteSession(ctx, sessionID); err != nil {
+			return false, err
+		}
+		return true, nil
+	}
 	if b.sessionStore == nil {
 		return false, fmt.Errorf("gateway runtime bridge: session store is unavailable")
 	}
@@ -670,6 +696,9 @@ func (b *gatewayRuntimePortBridge) RenameSession(ctx context.Context, input gate
 	if title == "" {
 		return fmt.Errorf("gateway runtime bridge: title is required for rename")
 	}
+	if renamer, ok := b.runtime.(runtimeSessionRenamer); ok {
+		return renamer.RenameSession(ctx, sessionID, title)
+	}
 	if b.sessionStore == nil {
 		return fmt.Errorf("gateway runtime bridge: session store is unavailable")
 	}
@@ -919,6 +948,9 @@ func (b *gatewayRuntimePortBridge) SetSessionModel(ctx context.Context, input ga
 	if err != nil {
 		return err
 	}
+	if updater, ok := b.runtime.(runtimeSessionModelUpdater); ok {
+		return updater.UpdateSessionModel(ctx, session.ID, providerID, modelID)
+	}
 	head := session.HeadSnapshot()
 	head.Provider = providerID
 	head.Model = modelID
@@ -1999,6 +2031,9 @@ func (b *gatewayRuntimePortBridge) SyncSessionsProviderModel(
 	if providerID == "" || modelID == "" {
 		return nil
 	}
+	if syncer, ok := b.runtime.(runtimeSessionsProviderModelSyncer); ok {
+		return syncer.SyncSessionsProviderModel(ctx, providerID, modelID)
+	}
 
 	summaries, err := b.runtime.ListSessions(ctx)
 	if err != nil {

internal/cli/gateway_runtime_bridge_test.go

@@ -367,6 +367,46 @@ func (s *bridgeSessionStoreStub) UpdateSessionState(ctx context.Context, input a
 	return nil
 }
 
+type boundaryRuntimeStub struct {
+	*runtimeStub
+	deletedSessionID string
+	renamedSessionID string
+	renamedTitle     string
+	modelSessionID   string
+	modelProviderID  string
+	modelID          string
+	syncProviderID   string
+	syncModelID      string
+}
+
+func (s *boundaryRuntimeStub) SupportsSessionMutationBoundary() bool {
+	return true
+}
+
+func (s *boundaryRuntimeStub) DeleteSession(_ context.Context, sessionID string) error {
+	s.deletedSessionID = sessionID
+	return nil
+}
+
+func (s *boundaryRuntimeStub) RenameSession(_ context.Context, sessionID string, title string) error {
+	s.renamedSessionID = sessionID
+	s.renamedTitle = title
+	return nil
+}
+
+func (s *boundaryRuntimeStub) UpdateSessionModel(_ context.Context, sessionID string, providerID string, modelID string) error {
+	s.modelSessionID = sessionID
+	s.modelProviderID = providerID
+	s.modelID = modelID
+	return nil
+}
+
+func (s *boundaryRuntimeStub) SyncSessionsProviderModel(_ context.Context, providerID string, modelID string) error {
+	s.syncProviderID = providerID
+	s.syncModelID = modelID
+	return nil
+}
+
 func TestGatewayRuntimePortBridgeCheckpointOperations(t *testing.T) {
 	stub := &runtimeStub{
 		listCheckpointsResult: []agentsession.CheckpointRecord{
@@ -1496,6 +1536,32 @@ func TestGatewayRuntimePortBridgeDeleteSession(t *testing.T) {
 	})
 }
 
+func TestGatewayRuntimePortBridgeDeleteSessionUsesRuntimeBoundary(t *testing.T) {
+	store := &bridgeSessionStoreStub{
+		deleteFn: func(_ context.Context, _ string) error {
+			t.Fatalf("DeleteSession should use runtime boundary instead of direct store write")
+			return nil
+		},
+	}
+	stub := &boundaryRuntimeStub{runtimeStub: &runtimeStub{eventsCh: make(chan agentruntime.RuntimeEvent, 1)}}
+	bridge, err := newGatewayRuntimePortBridge(context.Background(), stub, store)
+	if err != nil {
+		t.Fatalf("new bridge: %v", err)
+	}
+	defer bridge.Close()
+
+	deleted, err := bridge.DeleteSession(context.Background(), gateway.DeleteSessionInput{
+		SubjectID: testBridgeSubjectID,
+		SessionID: " session-1 ",
+	})
+	if err != nil {
+		t.Fatalf("DeleteSession() error = %v", err)
+	}
+	if !deleted || stub.deletedSessionID != "session-1" {
+		t.Fatalf("deleted=%v runtimeID=%q", deleted, stub.deletedSessionID)
+	}
+}
+
 func TestGatewayRuntimePortBridgeRenameSession(t *testing.T) {
 	t.Run("success", func(t *testing.T) {
 		store := &bridgeSessionStoreStub{
@@ -1564,6 +1630,32 @@ func TestGatewayRuntimePortBridgeRenameSession(t *testing.T) {
 	})
 }
 
+func TestGatewayRuntimePortBridgeRenameSessionUsesRuntimeBoundary(t *testing.T) {
+	store := &bridgeSessionStoreStub{
+		updateFn: func(_ context.Context, _ agentsession.UpdateSessionStateInput) error {
+			t.Fatalf("RenameSession should use runtime boundary instead of direct store write")
+			return nil
+		},
+	}
+	stub := &boundaryRuntimeStub{runtimeStub: &runtimeStub{eventsCh: make(chan agentruntime.RuntimeEvent, 1)}}
+	bridge, err := newGatewayRuntimePortBridge(context.Background(), stub, store)
+	if err != nil {
+		t.Fatalf("new bridge: %v", err)
+	}
+	defer bridge.Close()
+
+	if err := bridge.RenameSession(context.Background(), gateway.RenameSessionInput{
+		SubjectID: testBridgeSubjectID,
+		SessionID: " session-1 ",
+		Title:     " New Title ",
+	}); err != nil {
+		t.Fatalf("RenameSession() error = %v", err)
+	}
+	if stub.renamedSessionID != "session-1" || stub.renamedTitle != "New Title" {
+		t.Fatalf("runtime rename = (%q, %q)", stub.renamedSessionID, stub.renamedTitle)
+	}
+}
+
 // ---- providerSelection stub ----
 
 type providerSelectionStub struct {
@@ -2099,6 +2191,58 @@ func TestGatewayRuntimePortBridgeSetSessionModelProviderInference(t *testing.T)
 	}
 }
 
+func TestGatewayRuntimePortBridgeSetSessionModelUsesRuntimeBoundary(t *testing.T) {
+	store := &bridgeSessionStoreWithLoader{
+		bridgeSessionStoreStub: bridgeSessionStoreStub{
+			updateFn: func(_ context.Context, _ agentsession.UpdateSessionStateInput) error {
+				t.Fatalf("SetSessionModel should use runtime boundary instead of direct store write")
+				return nil
+			},
+		},
+		session: agentsession.Session{ID: "s-1", Provider: "", Model: ""},
+	}
+	ps := &providerSelectionStub{
+		listOptions: []configstate.ProviderOption{
+			{ID: "openai", Models: []providertypes.ModelDescriptor{{ID: "gpt-4", Name: "GPT-4"}}},
+		},
+	}
+	stub := &boundaryRuntimeStub{runtimeStub: &runtimeStub{eventsCh: make(chan agentruntime.RuntimeEvent, 1)}}
+	bridge, _ := newGatewayRuntimePortBridge(context.Background(), stub, store, nil, ps)
+	defer bridge.Close()
+
+	if err := bridge.SetSessionModel(context.Background(), gateway.SetSessionModelInput{
+		SubjectID: testBridgeSubjectID,
+		SessionID: "s-1",
+		ModelID:   "gpt-4",
+	}); err != nil {
+		t.Fatalf("SetSessionModel() error = %v", err)
+	}
+	if stub.modelSessionID != "s-1" || stub.modelProviderID != "openai" || stub.modelID != "gpt-4" {
+		t.Fatalf("runtime model update = (%q, %q, %q)", stub.modelSessionID, stub.modelProviderID, stub.modelID)
+	}
+}
+
+func TestGatewayRuntimePortBridgeSyncSessionsProviderModelUsesRuntimeBoundary(t *testing.T) {
+	store := &bridgeSessionStoreWithLoader{
+		bridgeSessionStoreStub: bridgeSessionStoreStub{
+			updateFn: func(_ context.Context, _ agentsession.UpdateSessionStateInput) error {
+				t.Fatalf("SyncSessionsProviderModel should use runtime boundary instead of direct store write")
+				return nil
+			},
+		},
+	}
+	stub := &boundaryRuntimeStub{runtimeStub: &runtimeStub{eventsCh: make(chan agentruntime.RuntimeEvent, 1)}}
+	bridge, _ := newGatewayRuntimePortBridge(context.Background(), stub, store)
+	defer bridge.Close()
+
+	if err := bridge.SyncSessionsProviderModel(context.Background(), " openai ", " gpt-4 "); err != nil {
+		t.Fatalf("SyncSessionsProviderModel() error = %v", err)
+	}
+	if stub.syncProviderID != "openai" || stub.syncModelID != "gpt-4" {
+		t.Fatalf("runtime sync = (%q, %q)", stub.syncProviderID, stub.syncModelID)
+	}
+}
+
 func TestGatewayRuntimePortBridgeSetSessionModelNotFound(t *testing.T) {
 	store := &bridgeSessionStoreWithLoader{
 		session: agentsession.Session{ID: "s-1", Provider: "openai", Model: ""},

internal/repository/git.go

@@ -441,14 +441,14 @@ func isAmbiguousGitStatusOutsideRepo(workdir string, output string, err error) b
 func hasGitMetadataAncestor(workdir string) bool {
 	current := filepath.Clean(workdir)
 	for {
-		gitPath := filepath.Join(current, ".git")
-		if _, err := os.Stat(gitPath); err == nil {
-			return true
-		}
 		parent := filepath.Dir(current)
 		if parent == current {
 			return false
 		}
+		gitPath := filepath.Join(current, ".git")
+		if _, err := os.Stat(gitPath); err == nil {
+			return true
+		}
 		current = parent
 	}
 }

internal/runner/capability.go

@@ -97,7 +97,7 @@ func resolvePath(target string, workdir string) string {
 	if trimmed == "" {
 		return ""
 	}
-	if filepath.IsAbs(trimmed) {
+	if filepath.IsAbs(trimmed) || strings.HasPrefix(filepath.ToSlash(trimmed), "/") {
 		return trimmed
 	}
 	base := strings.TrimSpace(workdir)