Merge pull request #30 from phantom5099/fork-pr-337-1776410957

fix(runtime/session/config): resolve submit-event blocking and cancellation gaps

Author: phantom5099

internal/config/loader_test.go

@@ -1050,12 +1050,12 @@ func TestLoadCustomProvidersReadDirAndStatErrors(t *testing.T) {
 			t.Fatalf("WriteFile() error = %v", err)
 		}
 
-		_, err := loadCustomProviders(baseDir)
-		if err == nil {
-			t.Fatal("expected read providers dir error")
+		providers, err := loadCustomProviders(baseDir)
+		if err != nil {
+			t.Fatalf("expected read providers dir fallback, got %v", err)
 		}
-		if !strings.Contains(err.Error(), "read providers dir") {
-			t.Fatalf("expected read providers dir error, got %v", err)
+		if len(providers) != 0 {
+			t.Fatalf("expected empty providers on read fallback, got %d", len(providers))
 		}
 	})
 

internal/config/provider_loader.go

@@ -67,7 +67,7 @@ func loadCustomProviders(baseDir string) ([]ProviderConfig, error) {
 		if os.IsNotExist(err) {
 			return nil, nil
 		}
-		return nil, fmt.Errorf("config: read providers dir: %w", err)
+		return nil, nil
 	}
 
 	sort.Slice(entries, func(i, j int) bool {

internal/provider/openaicompat/chatcompletions/provider_test.go

@@ -123,8 +123,8 @@ func TestNewAndBuildRequest(t *testing.T) {
 		if downgradedSchema["type"] != "object" {
 			t.Fatalf("expected downgraded schema type object, got %+v", downgradedSchema["type"])
 		}
-		if downgradedSchema["x-neocode-schema-downgraded"] != true {
-			t.Fatalf("expected downgrade marker, got %+v", downgradedSchema)
+		if _, ok := downgradedSchema["x-neocode-schema-downgraded"]; ok {
+			t.Fatalf("expected no custom downgrade marker in outbound schema, got %+v", downgradedSchema)
 		}
 
 		withSessionAsset, err := BuildRequest(context.Background(), testCfg("https://api.example.com/v1", "gpt-4.1", "test-key"), providertypes.GenerateRequest{

internal/provider/openaicompat/chatcompletions/request.go

@@ -91,14 +91,10 @@ func normalizeToolSchemaForOpenAI(schema map[string]any) map[string]any {
 	typeName, _ := normalized["type"].(string)
 	if strings.TrimSpace(strings.ToLower(typeName)) != "object" {
 		normalized["type"] = "object"
-		normalized["x-neocode-schema-downgraded"] = true
 	}
 
 	if _, ok := normalized["properties"].(map[string]any); !ok {
 		normalized["properties"] = map[string]any{}
-		if strings.TrimSpace(strings.ToLower(typeName)) != "object" {
-			normalized["x-neocode-schema-downgraded"] = true
-		}
 	}
 	return normalized
 }

internal/runtime/input_prepare.go

@@ -5,10 +5,13 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	agentsession "neo-code/internal/session"
 )
 
+const prepareEventEmitTimeout = 200 * time.Millisecond
+
 // NewSessionInputPreparer 创建基于 session 子层实现的输入归一化适配器。
 func NewSessionInputPreparer(store agentsession.Store, assetStore agentsession.AssetStore) UserInputPreparer {
 	return sessionInputPreparer{
@@ -51,7 +54,7 @@ func (s *Service) PrepareUserInput(ctx context.Context, input PrepareInput) (Use
 	}
 
 	runID := strings.TrimSpace(input.RunID)
-	_ = s.emit(ctx, EventInputNormalized, runID, prepared.UserInput.SessionID, InputNormalizedPayload{
+	_ = s.emitPrepareEvent(ctx, EventInputNormalized, runID, prepared.UserInput.SessionID, InputNormalizedPayload{
 		TextLength: len([]rune(strings.TrimSpace(input.Text))),
 		ImageCount: len(input.Images),
 	})
@@ -60,7 +63,7 @@ func (s *Service) PrepareUserInput(ctx context.Context, input PrepareInput) (Use
 		if index >= 0 && index < len(input.Images) {
 			path = strings.TrimSpace(input.Images[index].Path)
 		}
-		_ = s.emit(ctx, EventAssetSaved, runID, prepared.UserInput.SessionID, AssetSavedPayload{
+		_ = s.emitPrepareEvent(ctx, EventAssetSaved, runID, prepared.UserInput.SessionID, AssetSavedPayload{
 			Index:    index,
 			Path:     path,
 			AssetID:  asset.ID,
@@ -86,13 +89,31 @@ func (s *Service) emitPrepareFailure(ctx context.Context, input PrepareInput, er
 		if session := strings.TrimSpace(saveErr.SessionID); session != "" {
 			sessionID = session
 		}
-		return s.emit(ctx, EventAssetSaveFailed, runID, sessionID, AssetSaveFailedPayload{
+		return s.emitPrepareEvent(ctx, EventAssetSaveFailed, runID, sessionID, AssetSaveFailedPayload{
 			Index:   saveErr.Index,
 			Path:    strings.TrimSpace(saveErr.Path),
 			Message: strings.TrimSpace(saveErr.Error()),
 		})
 	}
-	return s.emit(ctx, EventError, runID, sessionID, strings.TrimSpace(err.Error()))
+	return s.emitPrepareEvent(ctx, EventError, runID, sessionID, strings.TrimSpace(err.Error()))
+}
+
+// emitPrepareEvent 在输入归一化阶段使用限时上下文发事件，避免通道拥塞导致提交链路卡死。
+func (s *Service) emitPrepareEvent(ctx context.Context, kind EventType, runID string, sessionID string, payload any) error {
+	emitCtx := ctx
+	cancel := func() {}
+	if _, hasDeadline := emitCtx.Deadline(); !hasDeadline {
+		emitCtx, cancel = context.WithTimeout(emitCtx, prepareEventEmitTimeout)
+	}
+	defer cancel()
+
+	if err := s.emit(emitCtx, kind, runID, sessionID, payload); err != nil {
+		if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+			return nil
+		}
+		return err
+	}
+	return nil
 }
 
 type sessionInputPreparer struct {

internal/runtime/input_prepare_test.go

@@ -123,6 +123,31 @@ func TestServiceSubmitWithoutPreparerReturnsError(t *testing.T) {
 	}
 }
 
+func TestServicePrepareUserInputDoesNotBlockWhenPrepareEventQueueIsFull(t *testing.T) {
+	t.Parallel()
+
+	workdir := t.TempDir()
+	svc, _ := newPrepareTestService(t, workdir, true)
+	for index := 0; index < cap(svc.events); index++ {
+		svc.events <- RuntimeEvent{Type: EventToolChunk}
+	}
+
+	start := time.Now()
+	input, err := svc.PrepareUserInput(context.Background(), PrepareInput{
+		RunID: "run-prepare-full-queue",
+		Text:  "hello",
+	})
+	if err != nil {
+		t.Fatalf("PrepareUserInput() error = %v", err)
+	}
+	if input.SessionID == "" {
+		t.Fatalf("expected prepared session id")
+	}
+	if elapsed := time.Since(start); elapsed > time.Second {
+		t.Fatalf("PrepareUserInput() blocked too long with full event queue: %v", elapsed)
+	}
+}
+
 func newPrepareTestService(t *testing.T, workdir string, withPreparer bool) (*Service, *agentsession.JSONStore) {
 	t.Helper()
 

internal/session/asset_store_test.go

@@ -118,6 +118,21 @@ func TestJSONStoreAssetStoreRespectsCanceledContext(t *testing.T) {
 	}
 }
 
+func TestJSONStoreSaveAssetStopsWhenContextCanceledDuringCopy(t *testing.T) {
+	t.Parallel()
+
+	store := NewJSONStore(t.TempDir(), t.TempDir())
+	ctx, cancel := context.WithCancel(context.Background())
+	reader := &cancelAfterFirstReadReader{
+		cancel: cancel,
+		chunks: [][]byte{[]byte("chunk-1"), []byte("chunk-2")},
+	}
+
+	if _, err := store.SaveAsset(ctx, "session_ctx_cancel_during_copy", reader, "image/png"); !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context canceled during copy, got %v", err)
+	}
+}
+
 func TestJSONStoreSaveAssetRejectsOversizedPayload(t *testing.T) {
 	t.Parallel()
 
@@ -224,8 +239,52 @@ func TestJSONStoreOpenAndStatMissingStoredFiles(t *testing.T) {
 	}
 }
 
+func TestJSONStoreDeleteAsset(t *testing.T) {
+	t.Parallel()
+
+	store := NewJSONStore(t.TempDir(), t.TempDir())
+	sessionID := "session-delete-asset"
+	meta, err := store.SaveAsset(context.Background(), sessionID, strings.NewReader("img"), "image/png")
+	if err != nil {
+		t.Fatalf("save seed asset: %v", err)
+	}
+
+	if err := store.DeleteAsset(context.Background(), sessionID, meta.ID); err != nil {
+		t.Fatalf("DeleteAsset() error = %v", err)
+	}
+	if _, statErr := os.Stat(store.assetPath(sessionID, meta.ID)); !errors.Is(statErr, os.ErrNotExist) {
+		t.Fatalf("expected removed asset file, got %v", statErr)
+	}
+	if _, statErr := os.Stat(store.assetMetaPath(sessionID, meta.ID)); !errors.Is(statErr, os.ErrNotExist) {
+		t.Fatalf("expected removed asset meta file, got %v", statErr)
+	}
+
+	if err := store.DeleteAsset(context.Background(), sessionID, meta.ID); err != nil {
+		t.Fatalf("DeleteAsset() should ignore already deleted files, got %v", err)
+	}
+}
+
 type failingReader struct{}
 
 func (failingReader) Read(_ []byte) (int, error) {
 	return 0, errors.New("read failure")
 }
+
+type cancelAfterFirstReadReader struct {
+	cancel context.CancelFunc
+	chunks [][]byte
+	index  int
+}
+
+func (r *cancelAfterFirstReadReader) Read(p []byte) (int, error) {
+	if r.index >= len(r.chunks) {
+		return 0, io.EOF
+	}
+	chunk := r.chunks[r.index]
+	r.index++
+	n := copy(p, chunk)
+	if r.index == 1 && r.cancel != nil {
+		r.cancel()
+	}
+	return n, nil
+}