package agenttask import ( "context" "fmt" "reflect" "sort" ) type integrationCandidate struct { ProjectID ProjectID WorkUnitID WorkUnitID Workspace WorkspaceID Ordinal DispatchOrdinal } func (m *Manager) integratePending( ctx context.Context, active []ProjectID, ) (bool, error) { state, err := m.load(ctx) if err != nil { return false, err } var candidates []integrationCandidate for _, projectID := range active { project := state.Projects[projectID] for workID, work := range project.Works { if work.State == WorkStatePendingIntegration { candidates = append(candidates, integrationCandidate{ ProjectID: projectID, WorkUnitID: workID, Workspace: project.WorkspaceID, Ordinal: work.DispatchOrdinal, }) } } } sort.Slice(candidates, func(left, right int) bool { if candidates[left].Ordinal != candidates[right].Ordinal { return candidates[left].Ordinal < candidates[right].Ordinal } if candidates[left].ProjectID != candidates[right].ProjectID { return candidates[left].ProjectID < candidates[right].ProjectID } return candidates[left].WorkUnitID < candidates[right].WorkUnitID }) progressed := false for _, candidate := range candidates { claimed, err := m.claimIntegration(ctx, candidate.Workspace) if err != nil { return progressed, err } if !claimed { var cmdID CommandID var wfRev WorkflowRevision if proj, ok := state.Projects[candidate.ProjectID]; ok && proj.Intent != nil { cmdID = proj.Intent.CommandID wfRev = proj.Intent.WorkflowRevision } m.emit(ctx, Event{ Type: EventBlocked, ProjectID: candidate.ProjectID, WorkspaceID: candidate.Workspace, WorkUnitID: candidate.WorkUnitID, CommandID: cmdID, WorkflowRevision: wfRev, Detail: string(BlockerDuplicateWorkspaceCall), }) continue } err = m.integrateOne(ctx, candidate) m.releaseIntegration(context.WithoutCancel(ctx), candidate.Workspace) if err != nil { return progressed, err } progressed = true } return progressed, nil } func (m *Manager) integrateOne( ctx context.Context, candidate integrationCandidate, ) error { project, work, err := m.loadWork(ctx, candidate.ProjectID, candidate.WorkUnitID) if err != nil { return err } if work.State != WorkStatePendingIntegration || work.ChangeSet == nil { return nil } integrationAttempt := work.IntegrationAttempt if integrationAttempt == 0 { integrationAttempt = 1 } if err := m.changeWork(ctx, candidate.ProjectID, candidate.WorkUnitID, func(work *WorkRecord) error { if err := transitionWork(work, WorkStateIntegrating); err != nil { return err } work.IntegrationAttempt = integrationAttempt return nil }); err != nil { return err } project, work, err = m.loadWork(ctx, candidate.ProjectID, candidate.WorkUnitID) if err != nil { return err } request := IntegrationRequest{ Project: project, Work: work, ChangeSet: *work.ChangeSet, Ordinal: work.DispatchOrdinal, Attempt: integrationAttempt, IdempotencyKey: integrationKey( candidate.ProjectID, candidate.WorkUnitID, *work.ChangeSet, integrationAttempt, ), } result, integrateErr := m.integrator.Integrate(ctx, request) if integrateErr != nil { result = IntegrationResult{ ProjectID: candidate.ProjectID, WorkUnitID: candidate.WorkUnitID, ChangeSet: *work.ChangeSet, Ordinal: work.DispatchOrdinal, Attempt: integrationAttempt, Outcome: IntegrationOutcomeTerminalDeferred, Retained: true, Blocker: &Blocker{ Code: BlockerIntegrationFailed, Message: integrateErr.Error(), Retryable: true, }, } } if err := validateIntegrationResult(request, result); err != nil { result = IntegrationResult{ ProjectID: candidate.ProjectID, WorkUnitID: candidate.WorkUnitID, ChangeSet: *work.ChangeSet, Ordinal: work.DispatchOrdinal, Attempt: integrationAttempt, Outcome: IntegrationOutcomeTerminalDeferred, Retained: true, Blocker: &Blocker{Code: BlockerIntegrationFailed, Message: err.Error()}, } } var next WorkState switch result.Outcome { case IntegrationOutcomeIntegrated: next = WorkStateCompleted case IntegrationOutcomeTerminalDeferred: next = WorkStateTerminalDeferred default: return fmt.Errorf("agenttask: unsupported integration outcome %q", result.Outcome) } err = m.changeWork(ctx, candidate.ProjectID, candidate.WorkUnitID, func(work *WorkRecord) error { if err := transitionWork(work, next); err != nil { return err } work.Integration = &result work.Blocker = result.Blocker return nil }) if err != nil { return err } var cmdID CommandID var wfRev WorkflowRevision if project.Intent != nil { cmdID = project.Intent.CommandID wfRev = project.Intent.WorkflowRevision } m.emit(ctx, Event{ Type: EventIntegrationResult, ProjectID: candidate.ProjectID, WorkspaceID: candidate.Workspace, WorkUnitID: candidate.WorkUnitID, CommandID: cmdID, WorkflowRevision: wfRev, AttemptID: work.AttemptID, Ordinal: candidate.Ordinal, ChangeSetID: result.ChangeSet.ID, ChangeSetRevision: result.ChangeSet.Revision, IntegrationAttempt: result.Attempt, State: next, Detail: string(result.Outcome), }) return nil } func validateIntegrationResult(request IntegrationRequest, result IntegrationResult) error { if result.ProjectID != request.Project.ProjectID || result.WorkUnitID != request.Work.Unit.ID || result.Ordinal != request.Ordinal || result.Attempt != request.Attempt || !reflect.DeepEqual(result.ChangeSet, request.ChangeSet) { return fmt.Errorf("agenttask: integration result durable identity mismatch") } switch result.Outcome { case IntegrationOutcomeIntegrated: if result.Blocker != nil { return fmt.Errorf("agenttask: integrated result cannot contain a blocker") } case IntegrationOutcomeTerminalDeferred: if !result.Retained || result.Blocker == nil { return fmt.Errorf("agenttask: terminal-deferred integration must retain its change set and blocker") } default: return fmt.Errorf("agenttask: unknown integration outcome") } return nil }