From 6d25f3d4edaa322f94673affef69f25c982e6563 Mon Sep 17 00:00:00 2001 From: dwrz Date: Wed, 7 Oct 2026 00:23:08 +0000 Subject: [PATCH] Refactor job pkg --- internal/job/job.go | 620 +++++++++++++++++++++++++++++++-------- internal/job/job_test.go | 590 +++++++++++++++++++++++++++++++++++-- 2 files changed, 1063 insertions(+), 147 deletions(-) diff --git a/internal/job/job.go b/internal/job/job.go index 4ffc233..8a7f236 100644 --- a/internal/job/job.go +++ b/internal/job/job.go @@ -1,44 +1,63 @@ // 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. When the service fires a -// job, it constructs messages and calls the LLM agent loop to execute the -// task. +// 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 is the cron.Schedule derived from schedule during construction. cron cron.Schedule - // log is the structured logger used for job execution. - log *slog.Logger + // 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 - // model is the model ID to pass to the LLM client (without provider - // prefix, which is resolved at construction time). - model string - // maxIterations caps the agent loop iterations for this job. Zero means - // unlimited. + // log is the structured logger used for job execution. + log *slog.Logger + // maxIterations caps the agent loop iterations. Zero means unlimited. maxIterations int - // name uniquely identifies the job within the config. It is used for - // logging, scheduling, and lookup. + // 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. @@ -49,17 +68,57 @@ type Job struct { // 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 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 } -// New creates a new Job from a Config. -// log is the structured logger used for job execution. -// llmc is the LLM client used to execute the job. -// model is the model ID (without provider prefix) to pass to the client. -func New(cfg config.JobConfig, log *slog.Logger, llmc *llm.Client, model string) (*Job, error) { - // Validation is handled by config.Config.Validate(). +// 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( @@ -67,33 +126,298 @@ func New(cfg config.JobConfig, log *slog.Logger, llmc *llm.Client, model string) ) } - var timeout time.Duration - if cfg.Timeout != "" { - timeout, err = time.ParseDuration(cfg.Timeout) - if err != nil { - return nil, fmt.Errorf( - "invalid timeout %q: %w", cfg.Timeout, err, - ) - } - } - return &Job{ cron: sched, - log: log.With(slog.String("job", cfg.Name)), + db: db, llm: llmc, - model: model, + 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: timeout, + 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 } @@ -103,6 +427,140 @@ 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 } @@ -114,95 +572,3 @@ func (j *Job) Task() string { return j.task } // Timeout returns the job's timeout duration. func (j *Job) Timeout() time.Duration { return j.timeout } - -// Result holds the outcome of a job execution. -type Result struct { - // Name is the job name. - Name string - // Duration is how long the job took to run. - Duration time.Duration - // 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 - // Error is set if the job failed. - Error error -} - -// Run executes the job: builds messages from the job config, calls the LLM, -// and logs the result. The agent uses tools for any additional output. -// Uses j.log and j.llm internally. -func (j *Job) Run(ctx context.Context) Result { - start := time.Now() - log := j.log.With(slog.String("job", j.Name())) - log.Info("starting job") - - // Build messages from job config. - messages := []llm.Message{ - { - Role: llm.RoleUser, - ContentParts: []llm.ContentPart{ - { - Text: j.Task(), - Type: llm.ContentTypeText, - }, - }, - }, - } - - // Apply per-job timeout if set. - if j.Timeout() > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, j.Timeout()) - defer cancel() - } - - res, err := j.llm.Completions( - ctx, llm.CompletionsParams{ - Model: j.model, - SystemMessage: j.SystemMessage(), - MaxIterations: j.MaxIterations(), - Messages: messages, - }, - ) - - duration := time.Since(start) - if err != nil { - 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), - } - } - if len(res.Messages) == 0 { - log.ErrorContext( - ctx, - "job returned no messages", - slog.Duration("duration", duration), - ) - return Result{ - Name: j.Name(), - Duration: duration, - Error: fmt.Errorf( - "job %q: no response from LLM", j.Name(), - ), - } - } - - log.InfoContext( - ctx, - "job completed", - slog.Duration("duration", duration), - slog.Int("messages", len(res.Messages)), - ) - return Result{ - Name: j.Name(), - Duration: duration, - Messages: res.Messages, - } -} diff --git a/internal/job/job_test.go b/internal/job/job_test.go index 74f910a..3d0ee0a 100644 --- a/internal/job/job_test.go +++ b/internal/job/job_test.go @@ -1,13 +1,52 @@ package job import ( + "database/sql" + "encoding/json/v2" + "fmt" + "io" "log/slog" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" "testing" "time" "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" ) +func testDependencies(t *testing.T) (*db.SQLite, *llm.Client, *tool.Registry) { + t.Helper() + + database, err := db.Connect(t.Context(), db.Config{ + Path: filepath.Join(t.TempDir(), "job.db"), + }) + if err != nil { + t.Fatalf("db.Connect: %v", err) + } + t.Cleanup(func() { database.Close() }) + if err := database.ApplySchema(t.Context(), slog.Default()); err != nil { + t.Fatalf("ApplySchema: %v", err) + } + client, err := llm.NewClient( + llm.Config{URL: "http://localhost"}, slog.Default(), + ) + if err != nil { + t.Fatalf("llm.NewClient: %v", err) + } + registry, err := tool.NewRegistry(nil) + if err != nil { + t.Fatalf("tool.NewRegistry: %v", err) + } + return database, client, registry +} + func TestConfigValidate(t *testing.T) { tests := []struct { name string @@ -24,6 +63,7 @@ func TestConfigValidate(t *testing.T) { config: config.JobConfig{ Schedule: "0 9 * * *", Task: "do something", + UserId: "default", }, wantErr: true, }, @@ -32,14 +72,25 @@ func TestConfigValidate(t *testing.T) { config: config.JobConfig{ Name: "test", Schedule: "0 9 * * *", + UserId: "default", }, wantErr: true, }, { name: "missing schedule", config: config.JobConfig{ - Name: "test", - Task: "do something", + Name: "test", + Task: "do something", + UserId: "default", + }, + wantErr: true, + }, + { + name: "missing user ID", + config: config.JobConfig{ + Name: "test", + Task: "do something", + Schedule: "0 9 * * *", }, wantErr: true, }, @@ -50,6 +101,7 @@ func TestConfigValidate(t *testing.T) { Task: "do something", Schedule: "0 9 * * *", MaxIterations: -1, + UserId: "default", }, wantErr: true, }, @@ -59,6 +111,7 @@ func TestConfigValidate(t *testing.T) { Name: "test", Task: "do something", Schedule: "0 9 * * *", + UserId: "default", }, wantErr: false, }, @@ -68,9 +121,10 @@ func TestConfigValidate(t *testing.T) { Name: "test", Task: "do something", Schedule: "0 9 * * *", - Timeout: "10m", + Timeout: 10 * time.Minute, MaxIterations: 20, SystemMessage: "You are an assistant.", + UserId: "default", }, wantErr: false, }, @@ -106,32 +160,28 @@ func TestConfigMaxIterations(t *testing.T) { } func TestConfigTimeout(t *testing.T) { - cfg := config.JobConfig{Name: "test"} - if cfg.Timeout != "" { - t.Errorf("default timeout = %q, want empty", cfg.Timeout) - } - - cfg.Timeout = "10m" - got, err := time.ParseDuration(cfg.Timeout) - if err != nil { - t.Fatalf("ParseDuration(%q) error = %v", cfg.Timeout, err) - } - if got.Minutes() != 10 { - t.Errorf("parsed timeout = %v, want 10m", got) + // A zero Timeout means the job has no per-job execution limit; + // Run then relies on the callers context. + var cfg config.JobConfig + if cfg.Timeout != 0 { + t.Errorf("default timeout = %v, want 0", cfg.Timeout) } } func TestNew(t *testing.T) { + database, client, registry := testDependencies(t) cfg := config.JobConfig{ Name: "test", Task: "do something", Schedule: "0 9 * * *", - Timeout: "10m", + Timeout: 10 * time.Minute, MaxIterations: 20, SystemMessage: "You are an assistant.", } - j, err := New(cfg, slog.Default(), nil, "test-model") + cfg.Model.Model = "test-model" + cfg.UserId = "default" + j, err := New(cfg, database.Writer, client, slog.Default(), registry) if err != nil { t.Fatalf("New() error = %v", err) } @@ -163,28 +213,528 @@ func TestNew(t *testing.T) { } func TestNew_InvalidSchedule(t *testing.T) { + database, client, registry := testDependencies(t) cfg := config.JobConfig{ Name: "test", Task: "do something", Schedule: "not-a-cron", } - _, err := New(cfg, slog.Default(), nil, "test-model") + cfg.Model.Model = "test-model" + cfg.UserId = "default" + _, err := New(cfg, database.Writer, client, slog.Default(), registry) if err == nil { t.Fatal("New() expected error for invalid schedule") } } func TestNew_InvalidTimeout(t *testing.T) { + database, client, registry := testDependencies(t) cfg := config.JobConfig{ Name: "test", Task: "do something", Schedule: "0 9 * * *", - Timeout: "not-a-duration", + Timeout: -time.Minute, } - _, err := New(cfg, slog.Default(), nil, "test-model") + cfg.Model.Model = "test-model" + cfg.UserId = "default" + _, err := New(cfg, database.Writer, client, slog.Default(), registry) if err == nil { t.Fatal("New() expected error for invalid timeout") } } + +func TestNewRequiresDependencies(t *testing.T) { + database, client, registry := testDependencies(t) + cfg := config.JobConfig{ + Name: "test", + Task: "do something", + Schedule: "0 9 * * *", + } + for _, test := range []struct { + name string + registry *tool.Registry + database *sql.DB + client *llm.Client + logger *slog.Logger + userID string + }{ + { + name: "nil registry", + database: database.Writer, + client: client, + logger: slog.Default(), + userID: "default", + }, + { + name: "nil database", + registry: registry, + client: client, + logger: slog.Default(), + userID: "default", + }, + { + name: "empty user ID", + registry: registry, + database: database.Writer, + client: client, + logger: slog.Default(), + }, + { + name: "nil llm client", + registry: registry, + database: database.Writer, + logger: slog.Default(), + userID: "default", + }, + { + name: "nil logger", + registry: registry, + database: database.Writer, + client: client, + userID: "default", + }, + } { + t.Run(test.name, func(t *testing.T) { + cfg.UserId = test.userID + _, err := New( + cfg, test.database, test.client, test.logger, + test.registry, + ) + if err == nil { + t.Fatal("New() expected error") + } + }) + } +} + +func TestRunPersistsToolLoop(t *testing.T) { + var requests atomic.Int32 + inference := httptest.NewServer(http.HandlerFunc(func( + w http.ResponseWriter, r *http.Request, + ) { + var request llm.ChatRequest + if err := json.UnmarshalRead(r.Body, &request); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + switch requests.Add(1) { + case 1: + if len(request.Messages) != 2 || + request.Messages[0].Role != llm.RoleSystem || + request.Messages[1].Role != llm.RoleUser { + http.Error(w, "unexpected initial history", http.StatusBadRequest) + return + } + fmt.Fprint(w, `{ + "choices":[{ + "message":{ + "role":"assistant", + "tool_calls":[{ + "id":"call_test", + "type":"function", + "function":{"name":"check_db","arguments":"{}"} + }] + }, + "finish_reason":"tool_calls" + }] + }`) + case 2: + if len(request.Messages) != 4 || + request.Messages[2].Role != llm.RoleAssistant || + request.Messages[3].Role != llm.RoleTool || + request.Messages[3].ToolCallID != "call_test" { + http.Error(w, "unexpected tool history", http.StatusBadRequest) + return + } + fmt.Fprint(w, `{ + "choices":[{ + "message":{"role":"assistant","content":"finished"}, + "finish_reason":"stop" + }] + }`) + default: + http.Error(w, "unexpected inference request", http.StatusBadRequest) + } + })) + defer inference.Close() + + var ( + ctx = t.Context() + log = slog.New(slog.NewTextHandler(io.Discard, nil)) + databasePath = filepath.Join(t.TempDir(), "job.db") + ) + database, err := db.Connect(ctx, db.Config{Path: databasePath}) + if err != nil { + t.Fatalf("db.Connect: %v", err) + } + defer database.Close() + if err := database.ApplySchema(ctx, slog.Default()); err != nil { + t.Fatalf("ApplySchema: %v", err) + } + t.Setenv("ODIDERE_JOB_TEST_DATABASE", databasePath) + executable, err := os.Executable() + if err != nil { + t.Fatalf("os.Executable: %v", err) + } + registry, err := tool.NewRegistry([]config.ToolConfig{{ + Name: "check_db", + Description: "Verify job persistence", + Command: executable, + Arguments: []string{"-test.run=^TestJobPersistenceHelper$"}, + Parameters: map[string]any{"type": "object"}, + }}) + if err != nil { + t.Fatalf("tool.NewRegistry: %v", err) + } + client, err := llm.NewClient(llm.Config{ + Provider: "test", + URL: inference.URL, + }, log) + if err != nil { + t.Fatalf("llm.NewClient: %v", err) + } + job, err := New( + config.JobConfig{ + Name: "daily report", + Task: "prepare the report", + Schedule: "0 9 * * *", + SystemMessage: "system", + Model: config.ModelConfig{Model: "test-model"}, + UserId: "default", + }, + database.Writer, + client, + log, + registry, + ) + if err != nil { + t.Fatalf("New: %v", err) + } + + result := job.Run(ctx) + if result.Error != nil { + t.Fatalf("Run: %v", result.Error) + } + if result.ConversationID == "" { + t.Fatal("conversation ID is empty") + } + if requests.Load() != 2 { + t.Fatalf("inference requests = %d, want 2", requests.Load()) + } + + q := db.New(database.Reader) + conversation, err := q.SelectConversationId(ctx, result.ConversationID) + if err != nil { + t.Fatalf("SelectConversationId: %v", err) + } + if conversation.UserId != "default" { + t.Errorf("conversation user = %q, want default", conversation.UserId) + } + if conversation.Title == nil || *conversation.Title != "daily report" { + t.Errorf("conversation title = %v, want daily report", conversation.Title) + } + stored, err := q.SelectMessagesConversationId(ctx, result.ConversationID) + if err != nil { + t.Fatalf("SelectMessagesConversationId: %v", err) + } + wantRoles := []db.MessageRole{ + db.MessageRoleSystem, + db.MessageRoleUser, + db.MessageRoleAssistant, + db.MessageRoleTool, + db.MessageRoleAssistant, + } + if len(stored) != len(wantRoles) { + t.Fatalf("stored messages = %d, want %d", len(stored), len(wantRoles)) + } + for i, want := range wantRoles { + if stored[i].Sequence != i || stored[i].Role != want { + t.Errorf( + "stored message %d = sequence %d, role %q; want %d, %q", + i, stored[i].Sequence, stored[i].Role, i, want, + ) + } + } +} + +func TestJobPersistenceHelper(t *testing.T) { + path := os.Getenv("ODIDERE_JOB_TEST_DATABASE") + if path == "" { + return + } + database, err := sql.Open("sqlite3", "file:"+path+"?mode=ro") + if err != nil { + t.Fatalf("sql.Open: %v", err) + } + defer database.Close() + + var role string + if err := database.QueryRow( + "SELECT role FROM messages ORDER BY created_at DESC, sequence DESC LIMIT 1", + ).Scan(&role); err != nil { + t.Fatalf("select latest message: %v", err) + } + if role != string(db.MessageRoleAssistant) { + t.Fatalf("latest message role = %q, want assistant", role) + } +} + +func TestCallTool(t *testing.T) { + database, client, _ := testDependencies(t) + registry, err := tool.NewRegistry([]config.ToolConfig{ + { + Name: "finish", + Description: "finish", + Command: "sh", + Arguments: []string{"-c", "echo done"}, + }, + { + Name: "fail", + Description: "fail", + Command: "sh", + Arguments: []string{"-c", "echo oops; exit 3"}, + }, + { + Name: "missing", + Description: "missing", + Command: "nonexistent_binary_xyz", + }, + { + Name: "interrupt", + Description: "interrupt", + Command: "sh", + // exec replaces the shell, so the timeout kills the + // process that holds the output pipe. + Arguments: []string{"-c", "echo partial; exec sleep 10"}, + Timeout: 100 * time.Millisecond, + }, + }) + if err != nil { + t.Fatalf("tool.NewRegistry: %v", err) + } + log := slog.New(slog.DiscardHandler) + job, err := New( + config.JobConfig{ + Name: "call tool", + Task: "call a tool", + Schedule: "0 9 * * *", + UserId: "default", + }, + database.Writer, + client, + log, + registry, + ) + if err != nil { + t.Fatalf("New: %v", err) + } + conversation, err := job.createConversation(t.Context(), nil) + if err != nil { + t.Fatalf("createConversation: %v", err) + } + + tests := []struct { + name string + tool string + // want lists text the message must contain, in this order. + want []string + // warn reports whether the message must carry + // interruptedWarning. + warn bool + // status is the execution row's status + // after callTool returns. + status db.ExecutionStatus + }{ + { + name: "finished", + tool: "finish", + want: []string{"▸ stdout:\ndone\n", "┗ exit: 0"}, + status: db.ExecutionStatusCompleted, + }, + { + name: "non-zero exit", + tool: "fail", + want: []string{"▸ stdout:\noops\n", "┗ exit: 3"}, + status: db.ExecutionStatusCompleted, + }, + { + name: "never started", + tool: "missing", + want: []string{"executable file not found"}, + status: db.ExecutionStatusCompleted, + }, + { + name: "interrupted", + tool: "interrupt", + want: []string{ + "timed out", + interruptedWarning, + "▸ stdout:\npartial\n", + }, + warn: true, + status: db.ExecutionStatusRunning, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + call := llm.ToolCall{ + Function: llm.Function{Arguments: "{}", Name: tt.tool}, + ID: "call_" + tt.tool, + Type: llm.ToolTypeFunction, + } + assistant, err := job.insertMessage( + t.Context(), conversation.Id, llm.Message{ + Role: llm.RoleAssistant, + ToolCalls: []llm.ToolCall{call}, + }, + nil, + ) + if err != nil { + t.Fatalf("insertMessage: %v", err) + } + + message, err := job.callTool( + t.Context(), log, conversation.Id, assistant.Id, call, + ) + if err != nil { + t.Fatalf("callTool: %v", err) + } + if message.Role != llm.RoleTool || + message.ToolCallID != call.ID || + message.Name != tt.tool { + t.Errorf( + "message = role %q, tool call %q, name %q; "+ + "want %q, %q, %q", + message.Role, message.ToolCallID, message.Name, + llm.RoleTool, call.ID, tt.tool, + ) + } + content := message.ContentParts[0].Text + rest := content + for _, want := range tt.want { + i := strings.Index(rest, want) + if i < 0 { + t.Fatalf( + "content missing %q, or out of order:\n%s", + want, content, + ) + } + rest = rest[i+len(want):] + } + if got := strings.Contains( + content, interruptedWarning, + ); got != tt.warn { + t.Errorf( + "content has warning = %t, want %t:\n%s", + got, tt.warn, content, + ) + } + + stored, err := db.New(database.Reader). + SelectMessagesConversationId( + t.Context(), conversation.Id, + ) + if err != nil { + t.Fatalf("SelectMessagesConversationId: %v", err) + } + want, err := json.Marshal(*message) + if err != nil { + t.Fatalf("json.Marshal: %v", err) + } + last := stored[len(stored)-1] + if last.Role != db.MessageRoleTool || + last.Content != string(want) { + t.Errorf( + "last stored message = role %q, %s; "+ + "want %q, %s", + last.Role, last.Content, + db.MessageRoleTool, want, + ) + } + + executions, err := db.New(database.Reader). + SelectExecutionsMessageId(t.Context(), assistant.Id) + if err != nil { + t.Fatalf("SelectExecutionsMessageId: %v", err) + } + if len(executions) != 1 || + executions[0].ToolCallId != call.ID || + executions[0].Status != tt.status { + t.Errorf( + "executions = %+v, want one %q row for %q", + executions, tt.status, call.ID, + ) + } + }) + } +} + +// TestCallToolUnsavedRow checks that callTool does not run the tool when +// its executions row cannot be saved. The row references the assistant +// message, so a message ID that does not exist makes the save fail. +func TestCallToolUnsavedRow(t *testing.T) { + database, client, _ := testDependencies(t) + marker := filepath.Join(t.TempDir(), "ran") + t.Setenv("ODIDERE_CALL_TOOL_MARKER", marker) + registry, err := tool.NewRegistry([]config.ToolConfig{{ + Name: "touch", + Description: "touch", + Command: "sh", + Arguments: []string{"-c", `touch "$ODIDERE_CALL_TOOL_MARKER"`}, + }}) + if err != nil { + t.Fatalf("tool.NewRegistry: %v", err) + } + log := slog.New(slog.DiscardHandler) + job, err := New( + config.JobConfig{ + Name: "unsaved row", + Task: "call a tool", + Schedule: "0 9 * * *", + UserId: "default", + }, + database.Writer, + client, + log, + registry, + ) + if err != nil { + t.Fatalf("New: %v", err) + } + + conversation, err := job.createConversation(t.Context(), nil) + if err != nil { + t.Fatalf("createConversation: %v", err) + } + + message, err := job.callTool( + t.Context(), log, conversation.Id, "missing", + llm.ToolCall{ + Function: llm.Function{ + Arguments: "{}", Name: "touch", + }, + ID: "call_touch", + Type: llm.ToolTypeFunction, + }, + ) + if err == nil { + t.Fatal("callTool: expected error") + } + if message != nil { + t.Errorf("message = %+v, want nil", message) + } + if _, err := os.Stat(marker); err == nil { + t.Error("tool ran without a saved row") + } + executions, err := db.New(database.Reader). + SelectExecutionsMessageId(t.Context(), "missing") + if err != nil { + t.Fatalf("SelectExecutionsMessageId: %v", err) + } + if len(executions) != 0 { + t.Errorf("executions = %+v, want none", executions) + } +}