Skip to content

Commit e301076

Browse files
gusye1234gusye1234
committed
fix: session learning status always pending
1 parent 455c477 commit e301076

7 files changed

Lines changed: 391 additions & 5 deletions

File tree

‎src/server/api/go/internal/bootstrap/container.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -309,9 +309,11 @@ func BuildContainer() *do.Injector {
309309
do.MustInvoke[repo.LearningSpaceSessionRepo](i),
310310
do.MustInvoke[repo.AgentSkillsRepo](i),
311311
do.MustInvoke[repo.SessionRepo](i),
312+
do.MustInvoke[repo.TaskRepo](i),
312313
do.MustInvoke[service.AgentSkillsService](i),
313314
do.MustInvoke[service.ArtifactService](i),
314315
configs.SkillTemplatesFS,
316+
do.MustInvoke[*zap.Logger](i),
315317
), nil
316318
})
317319

‎src/server/api/go/internal/modules/repo/learning_space.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,7 @@ type LearningSpaceSessionRepo interface {
170170
ExistsBySessionID(ctx context.Context, sessionID uuid.UUID) (bool, error)
171171
GetBySpaceAndSessionID(ctx context.Context, learningSpaceID, sessionID uuid.UUID) (*model.LearningSpaceSession, error)
172172
ListBySpaceID(ctx context.Context, learningSpaceID uuid.UUID) ([]*model.LearningSpaceSession, error)
173+
UpdateStatus(ctx context.Context, learningSpaceID, sessionID uuid.UUID, status string) error
173174
}
174175

175176
type learningSpaceSessionRepo struct {
@@ -211,3 +212,10 @@ func (r *learningSpaceSessionRepo) ListBySpaceID(ctx context.Context, learningSp
211212
Find(&items).Error
212213
return items, err
213214
}
215+
216+
func (r *learningSpaceSessionRepo) UpdateStatus(ctx context.Context, learningSpaceID, sessionID uuid.UUID, status string) error {
217+
return r.db.WithContext(ctx).
218+
Model(&model.LearningSpaceSession{}).
219+
Where("learning_space_id = ? AND session_id = ?", learningSpaceID, sessionID).
220+
Update("status", status).Error
221+
}

‎src/server/api/go/internal/modules/repo/session.go‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,8 @@ type SessionRepo interface {
3939
GetMessageByID(ctx context.Context, sessionID uuid.UUID, messageID uuid.UUID) (*model.Message, error)
4040
UpdateMessageMeta(ctx context.Context, messageID uuid.UUID, meta datatypes.JSONType[map[string]interface{}]) error
4141
CopySession(ctx context.Context, sessionID uuid.UUID) (*CopySessionResult, error)
42+
HasUnfinishedMessages(ctx context.Context, sessionID uuid.UUID) (bool, error)
43+
HasFailedMessages(ctx context.Context, sessionID uuid.UUID) (bool, error)
4244
}
4345

4446
// CopySessionResult contains the result of a copy operation
@@ -601,3 +603,21 @@ func (r *sessionRepo) CopySession(ctx context.Context, sessionID uuid.UUID) (*Co
601603

602604
return &result, nil
603605
}
606+
607+
func (r *sessionRepo) HasUnfinishedMessages(ctx context.Context, sessionID uuid.UUID) (bool, error) {
608+
var exists bool
609+
err := r.db.WithContext(ctx).Raw(
610+
"SELECT EXISTS(SELECT 1 FROM messages WHERE session_id = ? AND session_task_process_status IN ('pending', 'running'))",
611+
sessionID,
612+
).Scan(&exists).Error
613+
return exists, err
614+
}
615+
616+
func (r *sessionRepo) HasFailedMessages(ctx context.Context, sessionID uuid.UUID) (bool, error) {
617+
var exists bool
618+
err := r.db.WithContext(ctx).Raw(
619+
"SELECT EXISTS(SELECT 1 FROM messages WHERE session_id = ? AND session_task_process_status = 'failed')",
620+
sessionID,
621+
).Scan(&exists).Error
622+
return exists, err
623+
}

‎src/server/api/go/internal/modules/repo/task.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111

1212
type TaskRepo interface {
1313
ListBySessionWithCursor(ctx context.Context, sessionID uuid.UUID, afterCreatedAt time.Time, afterID uuid.UUID, limit int, timeDesc bool) ([]model.Task, error)
14+
HasSuccessTask(ctx context.Context, sessionID uuid.UUID) (bool, error)
1415
}
1516

1617
type taskRepo struct{ db *gorm.DB }
@@ -44,3 +45,12 @@ func (r *taskRepo) ListBySessionWithCursor(ctx context.Context, sessionID uuid.U
4445
var items []model.Task
4546
return items, q.Order(orderBy).Limit(limit).Find(&items).Error
4647
}
48+
49+
func (r *taskRepo) HasSuccessTask(ctx context.Context, sessionID uuid.UUID) (bool, error) {
50+
var exists bool
51+
err := r.db.WithContext(ctx).Raw(
52+
"SELECT EXISTS(SELECT 1 FROM tasks WHERE session_id = ? AND status = 'success')",
53+
sessionID,
54+
).Scan(&exists).Error
55+
return exists, err
56+
}

‎src/server/api/go/internal/modules/service/learning_space.go‎

Lines changed: 61 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"github.com/memodb-io/Acontext/internal/modules/model"
1212
"github.com/memodb-io/Acontext/internal/modules/repo"
1313
"github.com/memodb-io/Acontext/internal/pkg/paging"
14+
"go.uber.org/zap"
1415
"gorm.io/gorm"
1516
)
1617

@@ -85,9 +86,11 @@ type learningSpaceService struct {
8586
lsSessRepo repo.LearningSpaceSessionRepo
8687
skillsRepo repo.AgentSkillsRepo
8788
sessionRepo repo.SessionRepo
89+
taskRepo repo.TaskRepo
8890
agentSkillsSvc AgentSkillsService
8991
artifactSvc ArtifactService
9092
templateFS fs.ReadFileFS
93+
logger *zap.Logger
9194
}
9295

9396
// DefaultSkillTemplatePaths lists the embedded template paths created alongside
@@ -104,19 +107,23 @@ func NewLearningSpaceService(
104107
lsSessRepo repo.LearningSpaceSessionRepo,
105108
skillsRepo repo.AgentSkillsRepo,
106109
sessionRepo repo.SessionRepo,
110+
taskRepo repo.TaskRepo,
107111
agentSkillsSvc AgentSkillsService,
108112
artifactSvc ArtifactService,
109113
templateFS fs.ReadFileFS,
114+
logger *zap.Logger,
110115
) LearningSpaceService {
111116
return &learningSpaceService{
112117
lsRepo: lsRepo,
113118
lsSkillRepo: lsSkillRepo,
114119
lsSessRepo: lsSessRepo,
115120
skillsRepo: skillsRepo,
116121
sessionRepo: sessionRepo,
122+
taskRepo: taskRepo,
117123
agentSkillsSvc: agentSkillsSvc,
118124
artifactSvc: artifactSvc,
119125
templateFS: templateFS,
126+
logger: logger,
120127
}
121128
}
122129

@@ -385,6 +392,7 @@ func (s *learningSpaceService) GetSession(ctx context.Context, projectID, learni
385392
}
386393
return nil, err
387394
}
395+
s.resolvePendingStatus(ctx, lss)
388396
return lss, nil
389397
}
390398

@@ -394,7 +402,59 @@ func (s *learningSpaceService) ListSessions(ctx context.Context, projectID, lear
394402
return nil, err
395403
}
396404

397-
return s.lsSessRepo.ListBySpaceID(ctx, learningSpaceID)
405+
sessions, err := s.lsSessRepo.ListBySpaceID(ctx, learningSpaceID)
406+
if err != nil {
407+
return nil, err
408+
}
409+
for _, lss := range sessions {
410+
s.resolvePendingStatus(ctx, lss)
411+
}
412+
return sessions, nil
413+
}
414+
415+
// resolvePendingStatus lazily resolves a "pending" learning session to a
416+
// terminal status when it's clear that no SkillLearnTask will ever arrive.
417+
// The resolved status is persisted to the DB so subsequent reads skip the check.
418+
func (s *learningSpaceService) resolvePendingStatus(ctx context.Context, lss *model.LearningSpaceSession) {
419+
if lss.Status != "pending" {
420+
return
421+
}
422+
423+
hasUnfinished, err := s.sessionRepo.HasUnfinishedMessages(ctx, lss.SessionID)
424+
if err != nil {
425+
s.logger.Warn("lazy status resolution: failed to check unfinished messages",
426+
zap.String("session_id", lss.SessionID.String()), zap.Error(err))
427+
return
428+
}
429+
if hasUnfinished {
430+
return
431+
}
432+
433+
hasSuccess, err := s.taskRepo.HasSuccessTask(ctx, lss.SessionID)
434+
if err != nil {
435+
s.logger.Warn("lazy status resolution: failed to check success tasks",
436+
zap.String("session_id", lss.SessionID.String()), zap.Error(err))
437+
return
438+
}
439+
if hasSuccess {
440+
return
441+
}
442+
443+
resolvedStatus := "completed"
444+
hasFailed, err := s.sessionRepo.HasFailedMessages(ctx, lss.SessionID)
445+
if err != nil {
446+
s.logger.Warn("lazy status resolution: failed to check failed messages, defaulting to completed",
447+
zap.String("session_id", lss.SessionID.String()), zap.Error(err))
448+
} else if hasFailed {
449+
resolvedStatus = "failed"
450+
}
451+
452+
if err := s.lsSessRepo.UpdateStatus(ctx, lss.LearningSpaceID, lss.SessionID, resolvedStatus); err != nil {
453+
s.logger.Warn("lazy status resolution: failed to persist resolved status",
454+
zap.String("session_id", lss.SessionID.String()),
455+
zap.String("resolved_status", resolvedStatus), zap.Error(err))
456+
}
457+
lss.Status = resolvedStatus
398458
}
399459

400460
func (s *learningSpaceService) ExcludeSkill(ctx context.Context, projectID, learningSpaceID, skillID uuid.UUID) error {

0 commit comments

Comments
 (0)