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

internal/codesessions/managed_agent_code_session.go

@@ -23,7 +23,6 @@ type ManagedAgentCreateInput struct {
 	PermissionMode             string
 	DangerouslySkipPermissions bool
 	Config                     json.RawMessage
-	InitialEvents              []json.RawMessage
 }
 
 // ManagedAgentCreateResult 只在创建链路内短暂携带两份明文凭证，调用方应立即交给
@@ -66,7 +65,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 +98,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 +125,52 @@ func (s *Service) CreateManagedAgentCodeSession(ctx context.Context, input Manag
 	}, nil
 }
 
+// activateManagedAgentCodeSession hands off only explicitly queued startup
+// messages, then activates while holding the same Session lock as send.
+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
+		}
+		inputs := make([]db.AppendCodeSessionEventInput, 0, len(items))
+		for _, item := range items {
+			if item.Event.EventType != "user.message" {
+				return errors.New("session event queue contains a non-user message")
+			}
+			payload, err := workerPayloadForPublicEvent(
+				codeSession.ExternalID,
+				item.Event.Payload,
+				item.Event.ProcessedAt,
+			)
+			if err != nil {
+				return err
+			}
+			inbound, err := newInboundEventInput(codeSession.ExternalID, payload, "public-session")
+			if err != nil {
+				return err
+			}
+			inputs = append(inputs, inbound)
+		}
+		activated, err := s.db.ActivateManagedAgentCodeSessionWithQueue(
+			ctx,
+			codeSession,
+			items,
+			inputs,
+		)
+		if err != nil {
+			return err
+		}
+		if activated {
+			return nil
+		}
+	}
+}
+
 // 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 {

internal/db/code_sessions.go

@@ -8,6 +8,8 @@ import (
 	"errors"
 	"strings"
 	"time"
+
+	"github.com/jmoiron/sqlx"
 )
 
 type CodeSession struct {
@@ -282,6 +284,138 @@ func (d *DB) CreateCodeSession(ctx context.Context, input CreateCodeSessionInput
 	})
 }
 
+// ActivateManagedAgentCodeSessionWithQueue atomically transfers the complete
+// startup queue to inbound, clears the temporary responsibility, and activates
+// the Code Session while holding the same Session lock as event sends.
+func (d *DB) ActivateManagedAgentCodeSessionWithQueue(
+	ctx context.Context,
+	codeSession CodeSession,
+	items []SessionEventQueueItem,
+	inputs []AppendCodeSessionEventInput,
+) (bool, error) {
+	if len(items) != len(inputs) {
+		return false, ErrInvalidState
+	}
+	tx, err := d.sql.BeginTxx(ctx, nil)
+	if err != nil {
+		return false, err
+	}
+	defer tx.Rollback()
+
+	session, err := getSessionSQLX(
+		ctx,
+		tx,
+		lockSessionForEventsQuery,
+		sessionLookupArguments(codeSession.WorkspaceID, codeSession.SessionExternalID),
+	)
+	if err != nil {
+		return false, err
+	}
+	current, err := getCodeSessionSQLX(ctx, tx, `
+		select `+codeSessionColumns()+`
+		from code_sessions
+		where organization_id = :organization_id
+			and workspace_id = :workspace_id
+			and external_id = :external_id
+			and session_id = :session_id
+			and status = 'initializing'
+			and deleted_at is null
+		for update
+	`, map[string]any{
+		"organization_id": codeSession.OrganizationID,
+		"workspace_id":    codeSession.WorkspaceID,
+		"external_id":     codeSession.ExternalID,
+		"session_id":      session.ID,
+	})
+	if err != nil {
+		return false, err
+	}
+	for _, item := range items {
+		if item.sessionUUID != session.UUID ||
+			item.Event.UUID != item.sessionEventUUID ||
+			item.Event.OrganizationID != session.OrganizationID ||
+			item.Event.WorkspaceID != session.WorkspaceID ||
+			item.Event.SessionID != session.ID ||
+			item.Event.SessionExternalID != session.ExternalID ||
+			item.Event.EventType != "user.message" {
+			return false, ErrInvalidState
+		}
+	}
+	queueRows, err := listSessionEventQueueIdentityRows(ctx, tx, session, true)
+	if err != nil {
+		return false, err
+	}
+	if !sessionEventQueueItemsMatch(queueRows, items) {
+		return false, nil
+	}
+
+	for _, input := range inputs {
+		inserted, duplicate, err := d.appendCodeSessionEventSQLXTx(ctx, tx, current, "inbound", input)
+		if err != nil {
+			return false, err
+		}
+		if duplicate && inserted.CodeSessionExternalID != current.ExternalID {
+			return false, ErrInvalidState
+		}
+		if !duplicate {
+			current.LastInboundSequenceNum = inserted.SequenceNum
+		}
+	}
+	deletedResult, err := namedExecContext(ctx, tx, `
+		delete from session_event_queue
+		where organization_id = :organization_id
+			and workspace_id = :workspace_id
+			and session_uuid = CAST(:session_uuid AS uuid)
+	`, map[string]any{
+		"organization_id": session.OrganizationID,
+		"workspace_id":    session.WorkspaceID,
+		"session_uuid":    session.UUID,
+	})
+	if err != nil {
+		return false, err
+	}
+	deleted, err := deletedResult.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+	if deleted != int64(len(items)) {
+		return false, ErrPreconditionFailed
+	}
+
+	result, err := namedExecContext(ctx, tx, `
+		update code_sessions
+		set status = 'active', updated_at = :now
+		where organization_id = :organization_id
+			and workspace_id = :workspace_id
+			and id = :id
+			and external_id = :external_id
+			and session_id = :session_id
+			and status = 'initializing'
+			and deleted_at is null
+	`, map[string]any{
+		"organization_id": current.OrganizationID,
+		"workspace_id":    current.WorkspaceID,
+		"id":              current.ID,
+		"external_id":     current.ExternalID,
+		"session_id":      session.ID,
+		"now":             time.Now().UTC(),
+	})
+	if err != nil {
+		return false, err
+	}
+	updated, err := result.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+	if updated != 1 {
+		return false, ErrInvalidState
+	}
+	if err := tx.Commit(); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
 // codeSessionCredentialContextSelect 查询 code session 的鉴权身份信息。
 // OAuth token 鉴权和 session-ingress JWT 签发都会使用这些信息。
 // JOIN 中同时校验 organization、workspace 和 session 的归属，防止跨租户查询。
@@ -826,13 +960,30 @@ func (d *DB) appendCodeSessionEvent(ctx context.Context, direction string, codeS
 	if err != nil {
 		return CodeSessionEvent{}, false, err
 	}
+	event, duplicate, err := d.appendCodeSessionEventSQLXTx(ctx, tx, session, direction, input)
+	if err != nil {
+		return CodeSessionEvent{}, false, err
+	}
+	if err := tx.Commit(); err != nil {
+		return CodeSessionEvent{}, false, err
+	}
+	return event, duplicate, nil
+}
+
+func (d *DB) appendCodeSessionEventSQLXTx(
+	ctx context.Context,
+	tx *sqlx.Tx,
+	session CodeSession,
+	direction string,
+	input AppendCodeSessionEventInput,
+) (CodeSessionEvent, bool, error) {
 	if input.RequiredWorkerEpoch != nil && session.CurrentWorkerEpoch != *input.RequiredWorkerEpoch {
 		return CodeSessionEvent{}, false, ErrWorkerEpochMismatch
 	}
 	if input.IdempotencyKey != "" {
 		existing, err := d.getCodeSessionEventTx(ctx, tx, direction, session.WorkspaceID, input.IdempotencyKey)
 		if err == nil {
-			return existing, true, tx.Commit()
+			return existing, true, nil
 		}
 		if !errors.Is(err, ErrNotFound) {
 			return CodeSessionEvent{}, false, err
@@ -854,7 +1005,10 @@ func (d *DB) appendCodeSessionEvent(ctx context.Context, direction string, codeS
 		deliveryStatus = "queued"
 	}
 
-	var event CodeSessionEvent
+	var (
+		event CodeSessionEvent
+		err   error
+	)
 	eventArguments := map[string]any{
 		"external_id":              input.ExternalID,
 		"organization_id":          session.OrganizationID,
@@ -915,9 +1069,6 @@ func (d *DB) appendCodeSessionEvent(ctx context.Context, direction string, codeS
 	}); err != nil {
 		return CodeSessionEvent{}, false, err
 	}
-	if err := tx.Commit(); err != nil {
-		return CodeSessionEvent{}, false, err
-	}
 	return event, false, nil
 }
 

internal/db/db.go

@@ -35,6 +35,8 @@ var (
 	ErrFileReferenceNotFound = errors.New("file reference not found")
 )
 
+var ErrSessionStartupMessageConflict = errors.New("session startup message conflict")
+
 type DB struct {
 	Pool *pgxpool.Pool
 	sql  *sqlx.DB

internal/db/deployments.go

@@ -282,6 +282,15 @@ func (d *DB) CreateManualDeploymentRun(ctx context.Context, input CreateManualDe
 	if err != nil {
 		return DeploymentRun{}, Session{}, SessionThread{}, nil, err
 	}
+	startup, err := sessionUserMessageStartupWindowSQLX(ctx, tx, session)
+	if err != nil {
+		return DeploymentRun{}, Session{}, SessionThread{}, nil, err
+	}
+	if startup {
+		if err := enqueueSessionEventsSQLXTx(ctx, tx, session, events); err != nil {
+			return DeploymentRun{}, Session{}, SessionThread{}, nil, err
+		}
+	}
 
 	run := input.Run
 	run.DeploymentID = deployment.ID