fix(sessions): 启动期用户消息可靠投递

internal/codesessions/managed_agent_code_session.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 	"time"
 
+	"github.com/samber/lo"
 	"github.com/superduck-ai/open-managed-agents/internal/auth"
 	"github.com/superduck-ai/open-managed-agents/internal/db"
 	"github.com/superduck-ai/open-managed-agents/internal/ids"
@@ -23,7 +25,6 @@ type ManagedAgentCreateInput struct {
 	PermissionMode             string
 	DangerouslySkipPermissions bool
 	Config                     json.RawMessage
-	InitialEvents              []json.RawMessage
 }
 
 // ManagedAgentCreateResult 只在创建链路内短暂携带两份明文凭证，调用方应立即交给
@@ -66,7 +67,7 @@ func (s *Service) CreateManagedAgentCodeSession(ctx context.Context, input Manag
 		WorkDir:               strings.TrimSpace(input.WorkDir),
 		PermissionMode:        strings.TrimSpace(input.PermissionMode),
 		Model:                 strings.TrimSpace(input.Model),
-		Status:                "active",
+		Status:                "initializing",
 		Metadata:              metadata,
 		// OAuth-compatible token 只落 SHA-256 hash；明文仅存在于当前返回值中。
 		OAuthAccessTokenHash: auth.HashAPIKey(oauthAccessToken),
@@ -99,7 +100,7 @@ func (s *Service) CreateManagedAgentCodeSession(ctx context.Context, input Manag
 	if err := s.queueInitialize(ctx, record, input.Config, now); err != nil {
 		return ManagedAgentCreateResult{}, err
 	}
-	if err := s.queueInitialPublicSessionEvents(ctx, record, input.InitialEvents, now); err != nil {
+	if err := s.activateManagedAgentCodeSession(ctx, input.Session, record); err != nil {
 		return ManagedAgentCreateResult{}, err
 	}
 	credentialContext, err := s.db.GetCodeSessionCredentialContextForIssue(
@@ -126,6 +127,188 @@ func (s *Service) CreateManagedAgentCodeSession(ctx context.Context, input Manag
 	}, nil
 }
 
+// activateManagedAgentCodeSession merges public session history with the
+// startup queue into inbound, then activates under the same Session lock as send.
+//
+// Order: historical forwardable session_events (excluding UUIDs still in the
+// queue) first, then queue items in FIFO order. Queue remains the sole source of
+// startup-window responsibility and cutover matching.
+func (s *Service) activateManagedAgentCodeSession(
+	ctx context.Context,
+	session db.Session,
+	codeSession db.CodeSession,
+) error {
+	for {
+		items, err := s.db.ListSessionEventQueueItems(ctx, session)
+		if err != nil {
+			return err
+		}
+		history, err := s.listSessionEventsAscending(ctx, session)
+		if err != nil {
+			return err
+		}
+		queuedUUIDs := lo.SliceToMap(items, func(item db.SessionEventQueueItem) (string, struct{}) {
+			return item.Event.UUID, struct{}{}
+		})
+		historyOnly := lo.Filter(history, func(event db.SessionEvent, _ int) bool {
+			_, queued := queuedUUIDs[event.UUID]
+			return !queued
+		})
+		historyInputs, err := s.inboundInputsFromPublicSessionEvents(codeSession.ExternalID, historyOnly)
+		if err != nil {
+			return err
+		}
+		queueInputs, err := lo.MapErr(items, func(item db.SessionEventQueueItem, _ int) (db.AppendCodeSessionEventInput, error) {
+			if item.Event.EventType != "user.message" {
+				return db.AppendCodeSessionEventInput{}, fmt.Errorf(
+					"%w: session event queue contains a non-user message",
+					db.ErrInvalidState,
+				)
+			}
+			return s.inboundInputFromPublicSessionEvent(codeSession.ExternalID, item.Event)
+		})
+		if err != nil {
+			return err
+		}
+		inputs := append(historyInputs, queueInputs...)
+		activated, err := s.ActivateManagedAgentCodeSessionWithQueue(
+			ctx,
+			codeSession,
+			items,
+			inputs,
+		)
+		if err != nil {
+			return err
+		}
+		if activated {
+			return nil
+		}
+	}
+}
+
+// ActivateManagedAgentCodeSessionWithQueue atomically writes startup inputs,
+// clears the matched queue snapshot, and activates the Code Session. The
+// service owns the cross-resource ordering; each transaction method owns only
+// its resource SQL.
+func (s *Service) ActivateManagedAgentCodeSessionWithQueue(
+	ctx context.Context,
+	codeSession db.CodeSession,
+	items []db.SessionEventQueueItem,
+	inputs []db.AppendCodeSessionEventInput,
+) (bool, error) {
+	if s == nil || s.db == nil {
+		return false, db.ErrNotFound
+	}
+	activated := false
+	err := s.db.WithManagedAgentActivationTx(ctx, func(tx db.ManagedAgentActivationTx) error {
+		session, err := tx.LockSessionForEvents(
+			ctx,
+			codeSession.WorkspaceUUID,
+			codeSession.SessionExternalID,
+		)
+		if err != nil {
+			return err
+		}
+		current, err := tx.LockInitializingCodeSession(ctx, codeSession.UUID)
+		if err != nil {
+			return err
+		}
+		if !lo.EveryBy(items, func(item db.SessionEventQueueItem) bool {
+			return item.Event.EventType == "user.message"
+		}) {
+			return db.ErrInvalidState
+		}
+		matches, err := tx.SessionEventQueueMatches(ctx, session, items)
+		if err != nil || !matches {
+			return err
+		}
+		for _, input := range inputs {
+			inserted, duplicate, err := tx.AppendCodeSessionInboundEvent(ctx, current, input)
+			if err != nil {
+				return err
+			}
+			if duplicate && inserted.CodeSessionExternalID != current.ExternalID {
+				return db.ErrInvalidState
+			}
+			if !duplicate {
+				current.LastInboundSequenceNum = inserted.SequenceNum
+			}
+		}
+		deleted, err := tx.DeleteSessionEventQueue(ctx, session.UUID)
+		if err != nil {
+			return err
+		}
+		if deleted != int64(len(items)) {
+			return db.ErrPreconditionFailed
+		}
+		updated, err := tx.ActivateCodeSession(ctx, current.UUID, time.Now().UTC())
+		if err != nil {
+			return err
+		}
+		if !updated {
+			return db.ErrInvalidState
+		}
+		activated = true
+		return nil
+	})
+	if err != nil {
+		return false, err
+	}
+	return activated, nil
+}
+
+func (s *Service) listSessionEventsAscending(ctx context.Context, session db.Session) ([]db.SessionEvent, error) {
+	var out []db.SessionEvent
+	var cursor *db.SessionEventPageCursor
+	for {
+		events, hasMore, err := s.db.ListSessionEventsPage(ctx, db.ListSessionEventsPageParams{
+			WorkspaceUUID:     session.WorkspaceUUID,
+			SessionExternalID: session.ExternalID,
+			Limit:             100,
+			Cursor:            cursor,
+			Order:             "asc",
+		})
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, events...)
+		if !hasMore || len(events) == 0 {
+			return out, nil
+		}
+		last := events[len(events)-1]
+		cursor = &db.SessionEventPageCursor{CreatedAt: last.CreatedAt, UUID: last.UUID}
+	}
+}
+
+func (s *Service) inboundInputsFromPublicSessionEvents(
+	codeSessionID string,
+	events []db.SessionEvent,
+) ([]db.AppendCodeSessionEventInput, error) {
+	inputs := make([]db.AppendCodeSessionEventInput, 0, len(events))
+	for _, event := range events {
+		if !forwardPublicEventToWorker(event.EventType) {
+			continue
+		}
+		input, err := s.inboundInputFromPublicSessionEvent(codeSessionID, event)
+		if err != nil {
+			return nil, err
+		}
+		inputs = append(inputs, input)
+	}
+	return inputs, nil
+}
+
+func (s *Service) inboundInputFromPublicSessionEvent(
+	codeSessionID string,
+	event db.SessionEvent,
+) (db.AppendCodeSessionEventInput, error) {
+	payload, err := workerPayloadForPublicEvent(codeSessionID, event.Payload, event.ProcessedAt)
+	if err != nil {
+		return db.AppendCodeSessionEventInput{}, err
+	}
+	return newInboundEventInput(codeSessionID, payload, "public-session")
+}
+
 // TerminateManagedAgentCodeSession revokes a Code Session created for a
 // sandbox launch that failed before the runtime became usable.
 func (s *Service) TerminateManagedAgentCodeSession(

internal/codesessions/service.go

@@ -43,30 +43,6 @@ func NewServiceWithCredentials(database *db.DB, credentials *SessionCredentials,
 	return &Service{db: database, credentials: credentials, logger: logger}
 }
 
-func (s *Service) queueInitialPublicSessionEvents(ctx context.Context, codeSession db.CodeSession, payloads []json.RawMessage, now time.Time) error {
-	if len(payloads) == 0 {
-		return nil
-	}
-	workerPayloads := make([]json.RawMessage, 0, len(payloads))
-	for _, raw := range payloads {
-		object, err := decodeJSONObject(raw)
-		if err != nil {
-			s.logger.WarnContext(ctx, "skip initial code session event", "code_session_id", codeSession.ExternalID, "error", err)
-			continue
-		}
-		if !forwardPublicEventToWorker(stringField(object, "type")) {
-			continue
-		}
-		payload, err := workerPayloadForPublicEvent(codeSession.ExternalID, raw, now)
-		if err != nil {
-			s.logger.ErrorContext(ctx, "convert initial code session event", "code_session_id", codeSession.ExternalID, "error", err)
-			continue
-		}
-		workerPayloads = append(workerPayloads, payload)
-	}
-	return s.QueueRawPublicSessionEvents(ctx, codeSession, workerPayloads)
-}
-
 func (s *Service) QueuePublicSessionEvents(ctx context.Context, session db.Session, events []db.SessionEvent) error {
 	if s == nil || len(events) == 0 {
 		return nil
@@ -78,6 +54,9 @@ func (s *Service) QueuePublicSessionEvents(ctx context.Context, session db.Sessi
 		}
 		return err
 	}
+	if codeSession.Status != "active" {
+		return nil
+	}
 	payloads := make([]json.RawMessage, 0, len(events))
 	for _, event := range events {
 		if !forwardPublicEventToWorker(event.EventType) {
@@ -324,15 +303,23 @@ func (s *Service) queueInitialize(ctx context.Context, codeSession db.CodeSessio
 }
 
 func (s *Service) appendInboundPayload(ctx context.Context, codeSessionID string, payload json.RawMessage, source string) (db.CodeSessionEvent, bool, error) {
-	meta, err := BuildEventMetadata(codeSessionID, "inbound", payload)
+	input, err := newInboundEventInput(codeSessionID, payload, source)
 	if err != nil {
 		return db.CodeSessionEvent{}, false, err
 	}
+	return s.db.AppendCodeSessionInboundEvent(ctx, codeSessionID, input)
+}
+
+func newInboundEventInput(codeSessionID string, payload json.RawMessage, source string) (db.AppendCodeSessionEventInput, error) {
+	meta, err := BuildEventMetadata(codeSessionID, "inbound", payload)
+	if err != nil {
+		return db.AppendCodeSessionEventInput{}, err
+	}
 	eventID, err := ids.New("csev_")
 	if err != nil {
-		return db.CodeSessionEvent{}, false, err
+		return db.AppendCodeSessionEventInput{}, err
 	}
-	return s.db.AppendCodeSessionInboundEvent(ctx, codeSessionID, db.AppendCodeSessionEventInput{
+	return db.AppendCodeSessionEventInput{
 		ExternalID:     eventID,
 		EventType:      meta.EventType,
 		EventSubtype:   meta.EventSubtype,
@@ -344,7 +331,7 @@ func (s *Service) appendInboundPayload(ctx context.Context, codeSessionID string
 		DeliveryStatus: "queued",
 		Source:         strings.TrimSpace(source),
 		CreatedAt:      time.Now().UTC(),
-	})
+	}, nil
 }
 
 func (s *Service) publishPublicPayloads(ctx context.Context, codeSessionID string, payloads []json.RawMessage) error {