odidere/internal/job/job.go

// Package job provides cron-based scheduling for autonomous LLM tasks.
//
// Jobs are defined in the YAML configuration with a cron expression,
// a system message, and a task description. Each job owns its inference
// and tool-execution loop.
package job

import (
	"context"
	"database/sql"
	"encoding/json/v2"
	"errors"
	"fmt"
	"log/slog"
	"runtime"
	"time"

	"github.com/robfig/cron/v3"

	"code.chimeric.al/chimerical/odidere/internal/config"
	"code.chimeric.al/chimerical/odidere/internal/db"
	"code.chimeric.al/chimerical/odidere/internal/llm"
	"code.chimeric.al/chimerical/odidere/internal/tool"
)

const (
	// interruptedWarning indicates a tool call started but did not finish
	// normally, so its effects are unknown.
	interruptedWarning = `▸ WARNING: the tool started, but its outcome is unknown.
It may have had effects. Do not assume it is safe to repeat.`

	// resultSaveTimeout is the time limit for saving one tool result.
	resultSaveTimeout = time.Second
)

// errMaxIterations is the error Run reports when a job's
// MaxIterations limit is reached before the job completes.
// A job with MaxIterations set to zero has no iteration limit and never
// produces this error.
var errMaxIterations = errors.New("max iterations exceeded")

// Job defines a scheduled autonomous task.
//
// Job is immutable after construction via New. All fields are private and
// accessed through getter methods. This ensures that a Job's configuration
// cannot change once it is handed to the service.
type Job struct {
	// cron is the cron.Schedule derived from schedule during construction.
	cron cron.Schedule
	// db is the database connection pool used to persist job conversations.
	db *sql.DB
	// llm is the LLM client used to execute the job.
	llm *llm.Client
	// log is the structured logger used for job execution.
	log *slog.Logger
	// maxIterations caps the agent loop iterations. Zero means unlimited.
	maxIterations int
	// model is the model ID to pass to the LLM client.
	model string
	// name uniquely identifies the job within the config.
	name string
	// schedule is the original cron expression string from the config
	// (e.g. "0 9 * * *"). Stored for logging and display purposes.
	schedule string
	// systemMessage is the per-job system prompt that overrides the global
	// LLM system message. Empty string means use the default.
	systemMessage string
	// task is the user-level instruction passed to the LLM when the job
	// fires. This is the core of what the job does.
	task string
	// timeout is the parsed execution time limit.
	// Zero means no per-job timeout (falls back to the global LLM timeout).
	timeout time.Duration
	// tools contains the tools available to the job.
	tools *tool.Registry
	// userId identifies the owner of conversations created by the job.
	userId string
}

// Result holds the outcome of a job execution.
type Result struct {
	// ConversationID identifies the conversation for this invocation.
	ConversationID string
	// Duration is how long the job took to run.
	Duration time.Duration
	// Error is set if the job failed.
	Error error
	// Messages is the full conversation returned by the LLM agent loop,
	// including tool calls and tool results. Empty if the job failed.
	Messages []llm.Message
	// Name is the job name.
	Name string
}

// New constructs a Job from cfg and its runtime dependencies. New parses
// values required for execution and rejects dependencies that Run cannot
// operate without. log must be non-nil; pass slog.New(slog.DiscardHandler)
// to run without logs.
func New(
	cfg config.JobConfig,
	db *sql.DB,
	llmc *llm.Client,
	log *slog.Logger,
	tools *tool.Registry,
) (*Job, error) {
	if err := cfg.Validate(); err != nil {
		return nil, fmt.Errorf("invalid config: %w", err)
	}
	if db == nil {
		return nil, fmt.Errorf("database is nil")
	}
	if llmc == nil {
		return nil, fmt.Errorf("llm client is nil")
	}
	if log == nil {
		return nil, fmt.Errorf("logger is nil")
	}
	if tools == nil {
		return nil, fmt.Errorf("tools registry is nil")
	}

	sched, err := cron.ParseStandard(cfg.Schedule)
	if err != nil {
		return nil, fmt.Errorf(
			"invalid schedule %q: %w", cfg.Schedule, err,
		)
	}

	return &Job{
		cron:          sched,
		db:            db,
		llm:           llmc,
		log:           log.With(slog.String("job", cfg.Name)),
		maxIterations: cfg.MaxIterations,
		model:         cfg.Model.Model,
		name:          cfg.Name,
		schedule:      cfg.Schedule,
		systemMessage: cfg.SystemMessage,
		task:          cfg.Task,
		timeout:       cfg.Timeout,
		tools:         tools,
		userId:        cfg.UserId,
	}, nil
}

// callTool records one tool call as running, runs the tool, and saves
// the tool message that reports the outcome to the model.
//
// The record is a row in the executions table, saved before the tool
// starts. After an interruption, the row shows whether the tool may have
// run:
//   - No row: the tool never started. It is safe to run again.
//   - Row with status 'running': the tool may have run and had effects.
//   - Row with status 'completed': the tool finished or never started,
//     and the tool message is saved.
//
// Results:
//   - A message and a nil error: the message is saved. The message
//     reports the tool's outcome: exit code and output, the error that
//     kept the tool from starting, or the error that interrupted it.
//   - A nil message and an error: the outcome was not recorded. The tool
//     may have run and had effects.
//
// The message contains:
//   - If the tool finished: its output.
//   - If it never started: the error.
//   - If it started but was stopped, or its output was cut off: the
//     error, a warning that its effects are unknown, and any output.
//
// Limitations:
//   - callTool relies on [tool.Registry.Execute] returning a nil result
//     only when the process never started.
func (j *Job) callTool(
	ctx context.Context,
	log *slog.Logger,
	cid, mid string,
	call llm.ToolCall,
) (*llm.Message, error) {
	log = log.With(
		slog.String("tool_call_id", call.ID),
		slog.String("name", call.Function.Name),
	)
	execution, err := db.New(j.db).InsertExecution(ctx, db.InsertExecutionParams{
		MessageId:  mid,
		Status:     db.ExecutionStatusRunning,
		ToolCallId: call.ID,
	})
	if err != nil {
		return nil, fmt.Errorf(
			"insert execution for tool call %q: %w", call.ID, err,
		)
	}

	log.InfoContext(
		ctx,
		"calling tool",
		slog.String("args", call.Function.Arguments),
	)
	result, toolErr := j.tools.Execute(
		ctx, call.Function.Name, call.Function.Arguments,
	)
	var content string
	switch {
	case toolErr == nil:
		log.InfoContext(
			ctx,
			"called tool",
			slog.Duration(
				"duration",
				result.FinishedAt.Sub(result.StartedAt),
			),
		)
		content = result.FormatResult()
	case result == nil:
		log.ErrorContext(
			ctx,
			"tool did not start",
			slog.Any("error", toolErr),
		)
		content = toolErr.Error()
	default:
		log.ErrorContext(
			ctx,
			"tool interrupted",
			slog.Any("error", toolErr),
			slog.Duration(
				"duration",
				result.FinishedAt.Sub(result.StartedAt),
			),
		)
		content = toolErr.Error() + "\n" +
			interruptedWarning + "\n" +
			result.FormatResult()
	}
	message := llm.Message{
		Role: llm.RoleTool,
		ContentParts: []llm.ContentPart{{
			Type: llm.ContentTypeText,
			Text: content,
		}},
		Name:       call.Function.Name,
		ToolCallID: call.ID,
	}
	var update *db.UpdateExecutionStatusParams
	if toolErr == nil || result == nil {
		update = &db.UpdateExecutionStatusParams{
			Id:     execution.Id,
			Status: db.ExecutionStatusCompleted,
		}
	}
	// Try to persist the outcome even if the job was canceled after the
	// tool ran. If saving fails, the running row records that it may have run.
	saveCtx, saveCancel := context.WithTimeout(
		context.WithoutCancel(ctx), resultSaveTimeout,
	)
	defer saveCancel()
	if _, err := j.insertMessage(saveCtx, cid, message, update); err != nil {
		log.ErrorContext(
			ctx,
			"failed to persist tool result",
			slog.Any("error", err),
		)
		return nil, fmt.Errorf("persist tool call %q: %w", call.ID, err)
	}
	return &message, nil
}

// callTools runs every tool call in an assistant message and collects its
// persisted result. cid is the conversation's ID; mid is the assistant
// message's ID.
//
//   - Calls run concurrently, at most twice the number of CPUs at once.
//   - By design, each call saves its own result, so the saved order is
//     not fixed. The returned order may differ from the saved order.
//   - If a call's row or result cannot be saved, the other calls still run and
//     save.
//   - The returned error joins every failure. If it is non-nil, at least
//     one call has no result, so the next inference request cannot be
//     sent.
func (j *Job) callTools(
	ctx context.Context,
	log *slog.Logger,
	cid, mid string,
	message llm.Message,
) ([]llm.Message, error) {
	type outcome struct {
		err error
		msg *llm.Message
	}
	var (
		completed = make(chan outcome, len(message.ToolCalls))
		sem       = make(chan struct{}, runtime.NumCPU()*2)
	)
	for _, call := range message.ToolCalls {
		go func() {
			sem <- struct{}{}
			defer func() { <-sem }()
			msg, err := j.callTool(ctx, log, cid, mid, call)
			completed <- outcome{err: err, msg: msg}
		}()
	}

	var (
		errs    []error
		results []llm.Message
	)
	for range message.ToolCalls {
		result := <-completed
		if result.err != nil {
			log.ErrorContext(
				ctx,
				"tool call failed",
				slog.Any("error", result.err),
			)
			errs = append(errs, result.err)
			continue
		}
		results = append(results, *result.msg)
	}
	return results, errors.Join(errs...)
}

// createConversation creates the conversation for one run of the job.
//
// messages is the complete initial history, in the order it should appear.
// The conversation and its initial messages persist together.
func (j *Job) createConversation(
	ctx context.Context,
	messages []llm.Message,
) (*db.Conversation, error) {
	tx, err := j.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("begin conversation transaction: %w", err)
	}
	defer tx.Rollback()

	q := db.New(tx)
	c, err := q.InsertConversation(
		ctx,
		db.InsertConversationParams{
			Title:  &j.name,
			UserId: j.userId,
		},
	)
	if err != nil {
		return nil, fmt.Errorf("insert conversation: %w", err)
	}
	for sequence, message := range messages {
		content, err := json.Marshal(message)
		if err != nil {
			return nil, fmt.Errorf("marshal initial message: %w", err)
		}
		if _, err := q.InsertMessage(ctx, db.InsertMessageParams{
			Content:        string(content),
			ConversationId: c.Id,
			Role:           db.MessageRole(message.Role),
			Sequence:       sequence,
		}); err != nil {
			return nil, fmt.Errorf("insert initial message: %w", err)
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf(
			"commit conversation transaction: %w", err,
		)
	}
	return c, nil
}

// Cron returns the parsed cron schedule.
func (j *Job) Cron() cron.Schedule { return j.cron }

// insertMessage stores one message in a conversation, after every
// message already stored there.
// If executionUpdate is non-nil, its status change and the message commit in
// the same transaction. A nil update leaves execution statuses unchanged.
// The caller is responsible for identifying the execution and choosing a
// valid status transition.
func (j *Job) insertMessage(
	ctx context.Context,
	cid string,
	message llm.Message,
	executionUpdate *db.UpdateExecutionStatusParams,
) (*db.Message, error) {
	content, err := json.Marshal(message)
	if err != nil {
		return nil, fmt.Errorf("marshal message: %w", err)
	}

	tx, err := j.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("begin message transaction: %w", err)
	}
	defer tx.Rollback()

	q := db.New(tx)
	sequence, err := q.NextSequence(ctx, cid)
	if err != nil {
		return nil, fmt.Errorf("get next message sequence: %w", err)
	}
	m, err := q.InsertMessage(ctx, db.InsertMessageParams{
		Content:        string(content),
		ConversationId: cid,
		Role:           db.MessageRole(message.Role),
		Sequence:       sequence,
	})
	if err != nil {
		return nil, fmt.Errorf("insert message: %w", err)
	}
	if executionUpdate != nil {
		if err := q.UpdateExecutionStatus(ctx, *executionUpdate); err != nil {
			return nil, fmt.Errorf("update execution status: %w", err)
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("commit message transaction: %w", err)
	}
	return m, nil
}

// MaxIterations returns the configured max iterations.
func (j *Job) MaxIterations() int { return j.maxIterations }

// Model returns the model ID used by the job.
func (j *Job) Model() string { return j.model }

// Name returns the job name.
func (j *Job) Name() string { return j.name }

// Run executes the job's inference and tool-execution loop.
func (j *Job) Run(ctx context.Context) Result {
	var start = time.Now()
	j.log.InfoContext(ctx, "starting job")

	// Build messages from job config.
	var messages []llm.Message
	if j.SystemMessage() != "" {
		messages = append(messages, llm.Message{
			Role: llm.RoleSystem,
			ContentParts: []llm.ContentPart{
				{
					Text: j.SystemMessage(),
					Type: llm.ContentTypeText,
				},
			},
		})
	}
	messages = append(messages, llm.Message{
		Role: llm.RoleUser,
		ContentParts: []llm.ContentPart{
			{
				Text: j.Task(),
				Type: llm.ContentTypeText,
			},
		},
	})
	// Track the end of the initial messages.
	// Everything appended after is generated by the loop.
	base := len(messages)

	// Apply per-job timeout if set.
	if j.Timeout() > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, j.Timeout())
		defer cancel()
	}

	// Create a conversation for this run.
	c, err := j.createConversation(ctx, messages)
	if err != nil {
		duration := time.Since(start)
		j.log.ErrorContext(
			ctx,
			"job failed",
			slog.Any("error", err),
			slog.Duration("duration", duration),
		)
		return Result{
			Name:     j.Name(),
			Duration: duration,
			Error:    fmt.Errorf("job %q: %w", j.Name(), err),
		}
	}
	log := j.log.With(slog.String("conversation_id", c.Id))
	log.InfoContext(ctx, "created job conversation")

	// Run the inference and tool-execution loop.
	var completed bool
	for i := 0; j.MaxIterations() == 0 || i < j.MaxIterations(); i++ {
		var (
			result *llm.CompletionResult
			tm     *db.Message
		)
		// Request the next completion from the model.
		result, err = j.llm.Completions(
			ctx,
			llm.CompletionsParams{
				Model:    j.model,
				Messages: messages,
				Tools:    j.tools.ToAPI(),
			},
		)
		if err != nil {
			break
		}

		// Persist the assistant message before acting on it,
		// so an interruption leaves a committed record.
		message := result.Message
		tm, err = j.insertMessage(ctx, c.Id, message, nil)
		if err != nil {
			break
		}
		messages = append(messages, message)
		// No tool calls means the model is done.
		if len(message.ToolCalls) == 0 {
			completed = true
			break
		}

		// Execute every tool call in the batch, persisting each result.
		toolResults, toolErr := j.callTools(
			ctx, log, c.Id, tm.Id, message,
		)
		messages = append(messages, toolResults...)
		if toolErr != nil {
			err = toolErr
			break
		}
	}
	duration := time.Since(start)
	if err == nil && !completed {
		err = errMaxIterations
	}
	if err != nil {
		log.ErrorContext(
			ctx,
			"job failed",
			slog.Any("error", err),
			slog.Duration("duration", duration),
		)
		return Result{
			ConversationID: c.Id,
			Name:           j.Name(),
			Duration:       duration,
			Error:          fmt.Errorf("job %q: %w", j.Name(), err),
		}
	}

	log.InfoContext(
		ctx,
		"job completed",
		slog.Duration("duration", duration),
		slog.Int("messages", len(messages)-base),
	)
	return Result{
		ConversationID: c.Id,
		Name:           j.Name(),
		Duration:       duration,
		Messages:       messages[base:],
	}
}

// Schedule returns the job schedule string.
func (j *Job) Schedule() string { return j.schedule }

// SystemMessage returns the job's system message.
func (j *Job) SystemMessage() string { return j.systemMessage }

// Task returns the job's task description.
func (j *Job) Task() string { return j.task }

// Timeout returns the job's timeout duration.
func (j *Job) Timeout() time.Duration { return j.timeout }