<commit_before>package bot

import (
	"bufio"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"
)

var envPassThrough = []string{
	"HOME",
	"HOSTNAME",
	"LANG",
	"PATH",
	"USER",
}

// startPipeline is triggered by plugins, job triggers, scheduled tasks, and child jobs
// Called from dispatch: checkPluginMatchersAndRun,
// jobcommands: checkJobMatchersAndRun or ScheduledTask,
// runPipeline.
func (c *botContext) startPipeline(parent *botContext, t interface{}, ptype pipelineType, command string, args ...string) (ret TaskRetVal) {
	task, _, job := getTask(t)
	isJob := job != nil
	ppipeName := c.pipeName
	ppipeDesc := c.pipeDesc
	c.pipeName = task.name
	c.pipeDesc = task.Description
	// TODO: Replace the waitgroup, pluginsRunning, defer func(), etc.
	botCfg.Add(1)
	botCfg.Lock()
	botCfg.pluginsRunning++
	c.timeZone = botCfg.timeZone
	botCfg.Unlock()
	defer func() {
		botCfg.Lock()
		botCfg.pluginsRunning--
		// TODO: this check shouldn't be necessary; remove and test
		if botCfg.pluginsRunning >= 0 {
			botCfg.Done()
		}
		botCfg.Unlock()
	}()

	// redundant but explicit
	c.stage = primaryTasks
	// Once Active, we need to use the Mutex for access to some fields; see
	// botcontext/type botContext
	c.registerActive(nil)

	// A job is always the first task in a pipeline; a new sub-pipeline is created
	// if a job is added in another pipeline.
	if isJob {
		// TODO / NOTE: RawMsg will differ between plugins and triggers - document?
		c.jobName = task.name // Exclusive always uses the jobName, regardless of the task that calls it
		c.environment["GOPHER_JOB_NAME"] = c.jobName
		c.jobChannel = task.Channel
		botCfg.RLock()
		c.history = botCfg.history
		botCfg.RUnlock()
		c.workingDirectory = ""
		c.protected = task.Protected
		var jh jobHistory
		rememberRuns := job.HistoryLogs
		if rememberRuns == 0 {
			rememberRuns = 1
		}
		key := histPrefix + c.jobName
		tok, _, ret := checkoutDatum(key, &jh, true)
		if ret != Ok {
			Log(Error, fmt.Sprintf("Error checking out '%s', no history will be remembered for '%s'", key, c.pipeName))
		} else {
			var start time.Time
			if c.timeZone != nil {
				start = time.Now().In(c.timeZone)
			} else {
				start = time.Now()
			}
			c.runIndex = jh.NextIndex
			c.environment["GOPHER_RUN_INDEX"] = fmt.Sprintf("%d", c.runIndex)
			hist := historyLog{
				LogIndex:   c.runIndex,
				CreateTime: start.Format("Mon Jan 2 15:04:05 MST 2006"),
			}
			jh.NextIndex++
			jh.Histories = append(jh.Histories, hist)
			l := len(jh.Histories)
			if l > rememberRuns {
				jh.Histories = jh.Histories[l-rememberRuns:]
			}
			ret := updateDatum(key, tok, jh)
			if ret != Ok {
				Log(Error, fmt.Sprintf("Error updating '%s', no history will be remembered for '%s'", key, c.pipeName))
			} else {
				if job.HistoryLogs > 0 && c.history != nil {
					pipeHistory, err := c.history.NewHistory(c.jobName, hist.LogIndex, job.HistoryLogs)
					if err != nil {
						Log(Error, fmt.Sprintf("Error starting history for '%s', no history will be recorded: %v", c.pipeName, err))
					} else {
						c.logger = pipeHistory
					}
				} else {
					if c.history == nil {
						Log(Warn, "Error starting history, no history provider available")
					}
				}
			}
		}
		for _, p := range task.Parameters {
			_, exists := c.environment[p.Name]
			if !exists {
				c.environment[p.Name] = p.Value
			}
		}
		if !job.Quiet || c.verbose {
			r := c.makeRobot()
			iChannel := c.Channel // channel where job was triggered / run
			taskinfo := task.name
			if len(args) > 0 {
				taskinfo += " " + strings.Join(args, " ")
			}
			var link string
			if url, ok := c.history.GetHistoryURL(task.name, c.runIndex); ok {
				link = fmt.Sprintf(" (link: %s)", url)
			}
			switch ptype {
			case jobTrigger:
				r.SendChannelMessage(c.jobChannel, fmt.Sprintf("Starting job '%s', run %d%s - triggered by app '%s' in channel '%s'", taskinfo, c.runIndex, link, c.User, iChannel))
			case jobCmd:
				r.SendChannelMessage(c.jobChannel, fmt.Sprintf("Starting job '%s', run %d%s - requested by user '%s' in channel '%s'", taskinfo, c.runIndex, link, c.User, iChannel))
			case spawnedTask:
				r.SendChannelMessage(c.jobChannel, fmt.Sprintf("Starting job '%s', run %d%s - spawned by pipeline '%s': %s", taskinfo, c.runIndex, link, ppipeName, ppipeDesc))
			case scheduled:
				r.SendChannelMessage(c.jobChannel, fmt.Sprintf("Starting scheduled job '%s', run %d%s", taskinfo, c.runIndex, link))
			default:<commit_msg>Verify history exists before getting link<commit_after>package bot

import (
	"bufio"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"
)

var envPassThrough = []string{
	"HOME",
	"HOSTNAME",
	"LANG",
	"PATH",
	"USER",
}

// startPipeline is triggered by plugins, job triggers, scheduled tasks, and child jobs
// Called from dispatch: checkPluginMatchersAndRun,
// jobcommands: checkJobMatchersAndRun or ScheduledTask,
// runPipeline.
func (c *botContext) startPipeline(parent *botContext, t interface{}, ptype pipelineType, command string, args ...string) (ret TaskRetVal) {
	task, _, job := getTask(t)
	isJob := job != nil
	ppipeName := c.pipeName
	ppipeDesc := c.pipeDesc
	c.pipeName = task.name
	c.pipeDesc = task.Description
	// TODO: Replace the waitgroup, pluginsRunning, defer func(), etc.
	botCfg.Add(1)
	botCfg.Lock()
	botCfg.pluginsRunning++
	c.timeZone = botCfg.timeZone
	botCfg.Unlock()
	defer func() {
		botCfg.Lock()
		botCfg.pluginsRunning--
		// TODO: this check shouldn't be necessary; remove and test
		if botCfg.pluginsRunning >= 0 {
			botCfg.Done()
		}
		botCfg.Unlock()
	}()

	// redundant but explicit
	c.stage = primaryTasks
	// Once Active, we need to use the Mutex for access to some fields; see
	// botcontext/type botContext
	c.registerActive(nil)

	// A job is always the first task in a pipeline; a new sub-pipeline is created
	// if a job is added in another pipeline.
	if isJob {
		// TODO / NOTE: RawMsg will differ between plugins and triggers - document?
		c.jobName = task.name // Exclusive always uses the jobName, regardless of the task that calls it
		c.environment["GOPHER_JOB_NAME"] = c.jobName
		c.jobChannel = task.Channel
		botCfg.RLock()
		c.history = botCfg.history
		botCfg.RUnlock()
		c.workingDirectory = ""
		c.protected = task.Protected
		var jh jobHistory
		rememberRuns := job.HistoryLogs
		if rememberRuns == 0 {
			rememberRuns = 1
		}
		key := histPrefix + c.jobName
		tok, _, ret := checkoutDatum(key, &jh, true)
		if ret != Ok {
			Log(Error, fmt.Sprintf("Error checking out '%s', no history will be remembered for '%s'", key, c.pipeName))
		} else {
			var start time.Time
			if c.timeZone != nil {
				start = time.Now().In(c.timeZone)
			} else {
				start = time.Now()
			}
			c.runIndex = jh.NextIndex
			c.environment["GOPHER_RUN_INDEX"] = fmt.Sprintf("%d", c.runIndex)
			hist := historyLog{
				LogIndex:   c.runIndex,
				CreateTime: start.Format("Mon Jan 2 15:04:05 MST 2006"),
			}
			jh.NextIndex++
			jh.Histories = append(jh.Histories, hist)
			l := len(jh.Histories)
			if l > rememberRuns {
				jh.Histories = jh.Histories[l-rememberRuns:]
			}
			ret := updateDatum(key, tok, jh)
			if ret != Ok {
				Log(Error, fmt.Sprintf("Error updating '%s', no history will be remembered for '%s'", key, c.pipeName))
			} else {
				if job.HistoryLogs > 0 && c.history != nil {
					pipeHistory, err := c.history.NewHistory(c.jobName, hist.LogIndex, job.HistoryLogs)
					if err != nil {
						Log(Error, fmt.Sprintf("Error starting history for '%s', no history will be recorded: %v", c.pipeName, err))
					} else {
						c.logger = pipeHistory
					}
				} else {
					if c.history == nil {
						Log(Warn, "Error starting history, no history provider available")
					}
				}
			}
		}
		for _, p := range task.Parameters {
			_, exists := c.environment[p.Name]
			if !exists {
				c.environment[p.Name] = p.Value
			}
		}
		if !job.Quiet || c.verbose {
			r := c.makeRobot()
			iChannel := c.Channel // channel where job was triggered / run
			taskinfo := task.name
			if len(args) > 0 {
				taskinfo += " " + strings.Join(args, " ")
			}
			var link string
			if c.history != nil {
				if url, ok := c.history.GetHistoryURL(task.name, c.runIndex); ok {
					link = fmt.Sprintf(" (link: %s)", url)
				}
			}
			switch ptype {
			case jobTrigger:
				r.SendChannelMessage(c.jobChannel, fmt.Sprintf("Starting job '%s', run %d%s - triggered by app '%s' in channel '%s'", taskinfo, c.runIndex, link, c.User, iChannel))
			case jobCmd:
				r.SendChannelMessage(c.jobChannel, fmt.Sprintf("Starting job '%s', run %d%s - requested by user '%s' in channel '%s'", taskinfo, c.runIndex, link, c.User, iChannel))
			case spawnedTask:
				r.SendChannelMessage(c.jobChannel, fmt.Sprintf("Starting job '%s', run %d%s - spawned by pipeline '%s': %s", taskinfo, c.runIndex, link, ppipeName, ppipeDesc))
			case scheduled:
				r.SendChannelMessage(c.jobChannel, fmt.Sprintf("Starting scheduled job '%s', run %d%s", taskinfo, c.runIndex, link))
			default:
{'fraction_non_alphanumeric': 0.13755726896859985, 'fraction_numerical': 0.0037993071851603532, 'mean_word_length': 3.298751200768492, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2589141', 'n_tokens_mistral': 3481, 'n_tokens_neox': 3030, 'n_words': 877}