agent-radar
2ed3415fb7
Publish Claude Code Binaries / Stage and publish binaries (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-arm64 (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build linux-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build windows-amd64 (push) Canceled after 0s
Sync Tutti Agent Skills / sync (push) Canceled after 0s
Windows Agent Adapters / Test Windows agent adapters (push) Canceled after 0s
Windows Daemon Adapters / Test Windows daemon adapters (push) Canceled after 0s
Publish Tutti App Runtime / Publish runtime catalog (push) Canceled after 0s
754 lines
26 KiB
Go
754 lines
26 KiB
Go
package workspace
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
workspaceissues "github.com/tutti-os/tutti/packages/workspace/issues"
|
|
executionbiz "github.com/tutti-os/tutti/services/tuttid/biz/tuttimodeexecution"
|
|
)
|
|
|
|
func (s *SQLiteStore) AdmitTuttiModeSchedule(
|
|
ctx context.Context,
|
|
admission executionbiz.ScheduleAdmission,
|
|
) (executionbiz.ScheduleResult, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
tx, err := s.writeDB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return executionbiz.ScheduleResult{}, fmt.Errorf("begin Tutti mode schedule admission: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
replayed, found, err := getTuttiModeScheduleReplay(ctx, tx, admission)
|
|
if err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
if found {
|
|
return replayed, nil
|
|
}
|
|
|
|
executionID, err := validateTuttiModeScheduleFence(ctx, tx, admission)
|
|
if err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
if err := acceptTuttiModeSettlementSubjectForSchedule(
|
|
ctx, tx, executionID, admission,
|
|
); err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
tasks, issue, err := validateTuttiModeScheduleTasks(ctx, tx, admission)
|
|
if err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
if err := validateTuttiModeScheduleCapacity(ctx, tx, admission.WorkspaceID, issue, len(tasks)); err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
if err := insertTuttiModeScheduledRuns(ctx, tx, admission, tasks); err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
if err := projectWorkspaceIssueTasks(ctx, tx, admission.WorkspaceID, admission.IssueID, admission.Now); err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
if err := resolveTuttiModeScheduleCheckpoint(ctx, tx, executionID, admission); err != nil {
|
|
return executionbiz.ScheduleResult{}, err
|
|
}
|
|
|
|
result := executionbiz.ScheduleResult{
|
|
ExecutionID: executionID,
|
|
CheckpointID: admission.CheckpointID,
|
|
GraphRevision: admission.ExpectedGraphRevision,
|
|
RunIDs: make([]string, 0, len(admission.Runs)),
|
|
}
|
|
for _, run := range admission.Runs {
|
|
result.RunIDs = append(result.RunIDs, run.RunID)
|
|
}
|
|
resultJSON, err := json.Marshal(result)
|
|
if err != nil {
|
|
return executionbiz.ScheduleResult{}, fmt.Errorf("encode Tutti mode schedule result: %w", err)
|
|
}
|
|
_, err = tx.ExecContext(ctx, `
|
|
INSERT INTO workspace_tutti_execution_mutations (
|
|
workspace_id, execution_id, issue_id, source_session_id, kind, request_id,
|
|
input_sha256, checkpoint_id, expected_graph_revision, result_graph_revision,
|
|
result_json, created_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, 'schedule', ?, ?, ?, ?, ?, ?, ?)
|
|
`, admission.WorkspaceID, executionID, admission.IssueID, admission.SourceSessionID,
|
|
admission.RequestID, admission.InputSHA256, admission.CheckpointID,
|
|
admission.ExpectedGraphRevision, admission.ExpectedGraphRevision,
|
|
string(resultJSON), unixMs(admission.Now))
|
|
if err != nil {
|
|
if isSQLiteUniqueConstraintError(err) {
|
|
return executionbiz.ScheduleResult{}, executionbiz.ErrScheduleMutationConflict
|
|
}
|
|
return executionbiz.ScheduleResult{}, fmt.Errorf("record Tutti mode schedule mutation: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return executionbiz.ScheduleResult{}, fmt.Errorf("commit Tutti mode schedule admission: %w", err)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func acceptTuttiModeSettlementSubjectForSchedule(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
executionID string,
|
|
admission executionbiz.ScheduleAdmission,
|
|
) error {
|
|
var kind, subjectTaskID, subjectRunID string
|
|
err := tx.QueryRowContext(ctx, `
|
|
SELECT kind, subject_task_id, subject_run_id
|
|
FROM workspace_tutti_execution_checkpoints
|
|
WHERE workspace_id = ? AND execution_id = ? AND checkpoint_id = ?
|
|
`, admission.WorkspaceID, executionID, admission.CheckpointID).
|
|
Scan(&kind, &subjectTaskID, &subjectRunID)
|
|
if err != nil {
|
|
return fmt.Errorf("get scheduled settlement subject: %w", err)
|
|
}
|
|
if kind != string(executionbiz.CheckpointKindTaskSettled) {
|
|
return nil
|
|
}
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_issue_tasks
|
|
SET status = 'completed', acceptance_state = 'user_accepted',
|
|
updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ?
|
|
AND status = 'pending_acceptance'
|
|
AND EXISTS (
|
|
SELECT 1 FROM workspace_issue_runs
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ?
|
|
AND run_id = ? AND status = 'completed'
|
|
)
|
|
`, unixMs(admission.Now), admission.WorkspaceID, admission.IssueID, subjectTaskID,
|
|
admission.WorkspaceID, admission.IssueID, subjectTaskID, subjectRunID)
|
|
if err != nil {
|
|
return fmt.Errorf("accept settlement subject for schedule: %w", err)
|
|
}
|
|
return requireRowsAffected(
|
|
result, executionbiz.ErrScheduleRejected,
|
|
"accept settlement subject for schedule",
|
|
)
|
|
}
|
|
|
|
func getTuttiModeScheduleReplay(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
admission executionbiz.ScheduleAdmission,
|
|
) (executionbiz.ScheduleResult, bool, error) {
|
|
var inputSHA256, resultJSON string
|
|
err := tx.QueryRowContext(ctx, `
|
|
SELECT input_sha256, result_json
|
|
FROM workspace_tutti_execution_mutations
|
|
WHERE workspace_id = ? AND source_session_id = ? AND kind = 'schedule'
|
|
AND issue_id = ? AND request_id = ?
|
|
`, admission.WorkspaceID, admission.SourceSessionID, admission.IssueID, admission.RequestID).
|
|
Scan(&inputSHA256, &resultJSON)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return executionbiz.ScheduleResult{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return executionbiz.ScheduleResult{}, false, fmt.Errorf("get Tutti mode schedule replay: %w", err)
|
|
}
|
|
if inputSHA256 != admission.InputSHA256 {
|
|
return executionbiz.ScheduleResult{}, false, executionbiz.ErrScheduleMutationConflict
|
|
}
|
|
var result executionbiz.ScheduleResult
|
|
if err := json.Unmarshal([]byte(resultJSON), &result); err != nil {
|
|
return executionbiz.ScheduleResult{}, false, fmt.Errorf("decode Tutti mode schedule replay: %w", err)
|
|
}
|
|
result.Replayed = true
|
|
return result, true, nil
|
|
}
|
|
|
|
func validateTuttiModeScheduleFence(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
admission executionbiz.ScheduleAdmission,
|
|
) (string, error) {
|
|
var executionID, sourceSessionID, status string
|
|
var graphRevision int64
|
|
err := tx.QueryRowContext(ctx, `
|
|
SELECT execution_id, source_session_id, status, graph_revision
|
|
FROM workspace_tutti_executions
|
|
WHERE workspace_id = ? AND issue_id = ?
|
|
`, admission.WorkspaceID, admission.IssueID).
|
|
Scan(&executionID, &sourceSessionID, &status, &graphRevision)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", executionbiz.ErrExecutionNotFound
|
|
}
|
|
if err != nil {
|
|
return "", fmt.Errorf("get Tutti mode execution schedule fence: %w", err)
|
|
}
|
|
if sourceSessionID != admission.SourceSessionID {
|
|
return "", executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionWrongSourceSession, "",
|
|
)
|
|
}
|
|
if graphRevision != admission.ExpectedGraphRevision {
|
|
return "", executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionStaleGraphRevision, "",
|
|
)
|
|
}
|
|
if status != string(executionbiz.StatusAwaitingSchedule) &&
|
|
status != string(executionbiz.StatusAwaitingMain) {
|
|
return "", executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionInactiveExecution, "",
|
|
)
|
|
}
|
|
var checkpointStatus string
|
|
var checkpointRevision int64
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT status, graph_revision
|
|
FROM workspace_tutti_execution_checkpoints
|
|
WHERE workspace_id = ? AND execution_id = ? AND checkpoint_id = ?
|
|
`, admission.WorkspaceID, executionID, admission.CheckpointID).
|
|
Scan(&checkpointStatus, &checkpointRevision)
|
|
if err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected,
|
|
executionbiz.RejectionInactiveCheckpoint,
|
|
"",
|
|
)
|
|
}
|
|
return "", fmt.Errorf("get Tutti mode schedule checkpoint fence: %w", err)
|
|
}
|
|
if checkpointStatus != string(executionbiz.CheckpointStatusActive) {
|
|
return "", executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionInactiveCheckpoint, "",
|
|
)
|
|
}
|
|
if checkpointRevision != admission.ExpectedGraphRevision {
|
|
return "", executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionStaleGraphRevision, "",
|
|
)
|
|
}
|
|
return executionID, nil
|
|
}
|
|
|
|
func validateTuttiModeScheduleTasks(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
admission executionbiz.ScheduleAdmission,
|
|
) (map[string]workspaceissues.Task, workspaceissues.Issue, error) {
|
|
row := tx.QueryRowContext(ctx, fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issues
|
|
WHERE workspace_id = ? AND issue_id = ?
|
|
`, issueSelectColumns), admission.WorkspaceID, admission.IssueID)
|
|
issue, err := scanWorkspaceIssue(row)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil, workspaceissues.Issue{}, workspaceissues.ErrIssueNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, workspaceissues.Issue{}, fmt.Errorf("get scheduled Tutti mode Issue: %w", err)
|
|
}
|
|
if issue.PlanningSource != workspaceissues.PlanningSourceTuttiModePlan {
|
|
return nil, workspaceissues.Issue{}, executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionInactiveExecution, "",
|
|
)
|
|
}
|
|
if issue.SourceSessionID != admission.SourceSessionID {
|
|
return nil, workspaceissues.Issue{}, executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionWrongSourceSession, "",
|
|
)
|
|
}
|
|
if issue.DispatchPaused {
|
|
return nil, workspaceissues.Issue{}, executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionDispatchPaused, "",
|
|
)
|
|
}
|
|
if issue.Budget.Status != workspaceissues.BudgetStatusActive {
|
|
return nil, workspaceissues.Issue{}, executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionBudgetUnavailable, "",
|
|
)
|
|
}
|
|
|
|
rows, err := tx.QueryContext(ctx, fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issue_tasks
|
|
WHERE workspace_id = ? AND issue_id = ? AND superseded_at_unix_ms = 0
|
|
ORDER BY sort_index ASC, id ASC
|
|
`, taskSelectColumns), admission.WorkspaceID, admission.IssueID)
|
|
if err != nil {
|
|
return nil, workspaceissues.Issue{}, fmt.Errorf("list tasks for Tutti mode schedule: %w", err)
|
|
}
|
|
allTasks, err := scanWorkspaceIssueTasks(rows)
|
|
closeErr := rows.Close()
|
|
if err != nil {
|
|
return nil, workspaceissues.Issue{}, err
|
|
}
|
|
if closeErr != nil {
|
|
return nil, workspaceissues.Issue{}, fmt.Errorf("close tasks for Tutti mode schedule: %w", closeErr)
|
|
}
|
|
byID := make(map[string]workspaceissues.Task, len(allTasks))
|
|
for _, task := range allTasks {
|
|
byID[task.TaskID] = task
|
|
}
|
|
selected := make(map[string]workspaceissues.Task, len(admission.Runs))
|
|
for _, run := range admission.Runs {
|
|
if _, duplicate := selected[run.TaskID]; duplicate {
|
|
return nil, workspaceissues.Issue{}, executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionDuplicateTask, run.TaskID,
|
|
)
|
|
}
|
|
task, ok := byID[run.TaskID]
|
|
if !ok {
|
|
return nil, workspaceissues.Issue{}, executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionTaskNotFound, run.TaskID,
|
|
)
|
|
}
|
|
if blocker := workspaceissues.IssueTaskRunBlocker(task, byID); blocker != "" {
|
|
return nil, workspaceissues.Issue{}, executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected,
|
|
scheduleRejectionReasonForBlocker(blocker),
|
|
run.TaskID,
|
|
)
|
|
}
|
|
selected[run.TaskID] = task
|
|
}
|
|
return selected, issue, nil
|
|
}
|
|
|
|
func validateTuttiModeScheduleCapacity(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
workspaceID string,
|
|
issue workspaceissues.Issue,
|
|
requested int,
|
|
) error {
|
|
var running, activeIssueRuns int
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT COUNT(*), COALESCE(SUM(CASE WHEN issue_id = ? THEN 1 ELSE 0 END), 0)
|
|
FROM workspace_issue_runs
|
|
WHERE workspace_id = ? AND status = 'running' AND TRIM(agent_session_id) <> ''
|
|
`, issue.IssueID, workspaceID).Scan(&running, &activeIssueRuns); err != nil {
|
|
return fmt.Errorf("count running Issue Runs for Tutti mode schedule: %w", err)
|
|
}
|
|
if requested < 1 ||
|
|
requested > workspaceissues.IssueAutomaticRunAdmissionSlots(issue, running, activeIssueRuns) {
|
|
return executionbiz.Reject(
|
|
executionbiz.ErrScheduleRejected, executionbiz.RejectionCapacityExhausted, "",
|
|
)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func scheduleRejectionReasonForBlocker(
|
|
blocker workspaceissues.TaskRunBlocker,
|
|
) executionbiz.RejectionReason {
|
|
switch blocker {
|
|
case workspaceissues.TaskRunBlockerSuperseded:
|
|
return executionbiz.RejectionTaskSuperseded
|
|
case workspaceissues.TaskRunBlockerNotStarted:
|
|
return executionbiz.RejectionTaskNotStarted
|
|
case workspaceissues.TaskRunBlockerMissingAgentTarget:
|
|
return executionbiz.RejectionMissingAgentTarget
|
|
default:
|
|
return executionbiz.RejectionDependencyUnsatisfied
|
|
}
|
|
}
|
|
|
|
func insertTuttiModeScheduledRuns(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
admission executionbiz.ScheduleAdmission,
|
|
tasks map[string]workspaceissues.Task,
|
|
) error {
|
|
for _, run := range admission.Runs {
|
|
task := tasks[run.TaskID]
|
|
_, err := tx.ExecContext(ctx, `
|
|
INSERT INTO workspace_issue_runs (
|
|
run_id, task_id, issue_id, workspace_id, requester_user_id, agent_user_id,
|
|
agent_target_id, agent_session_id, agent_provider, model_plan_id, model,
|
|
reasoning_intensity, status, input_tokens, output_tokens, cache_read_tokens,
|
|
cache_write_tokens, summary, error_message, output_dir, execution_directory,
|
|
created_at_unix_ms, started_at_unix_ms, completed_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`, run.RunID, run.TaskID, run.IssueID, run.WorkspaceID, run.RequesterUserID,
|
|
run.AgentUserID, run.AgentTargetID, run.AgentSessionID, run.AgentProvider,
|
|
run.ModelPlanID, run.Model, run.ReasoningIntensity, string(run.Status),
|
|
run.Usage.InputTokens, run.Usage.OutputTokens, run.Usage.CacheReadTokens,
|
|
run.Usage.CacheWriteTokens, run.Summary, run.ErrorMessage, run.OutputDir,
|
|
run.ExecutionDirectory, run.CreatedAtUnixMS, run.StartedAtUnixMS,
|
|
run.CompletedAtUnixMS, run.UpdatedAtUnixMS)
|
|
if err != nil {
|
|
if isSQLiteUniqueConstraintError(err) {
|
|
return executionbiz.ErrScheduleRejected
|
|
}
|
|
return fmt.Errorf("create scheduled Tutti mode Run: %w", err)
|
|
}
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_issue_tasks
|
|
SET status = 'running', acceptance_state = 'agent_claimed',
|
|
acceptance_summary = '', latest_run_id = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ? AND status = 'not_started'
|
|
`, run.RunID, run.UpdatedAtUnixMS, admission.WorkspaceID, admission.IssueID, run.TaskID)
|
|
if err != nil {
|
|
return fmt.Errorf("claim scheduled Tutti mode task: %w", err)
|
|
}
|
|
if err := requireRowsAffected(result, executionbiz.ErrScheduleRejected, "claim scheduled Tutti mode task"); err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(ctx, `
|
|
INSERT INTO workspace_issue_run_launch_intents (
|
|
workspace_id, issue_id, task_id, run_id, launch_intent_id, client_submit_id,
|
|
status, created_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, 'prepared', ?, ?)
|
|
`, admission.WorkspaceID, admission.IssueID, task.TaskID, run.RunID,
|
|
"launch-intent:"+run.RunID, workspaceissues.IssueRunClientSubmitID(run.RunID),
|
|
run.CreatedAtUnixMS, run.UpdatedAtUnixMS)
|
|
if err != nil {
|
|
return fmt.Errorf("create scheduled Tutti mode launch intent: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func projectWorkspaceIssueTasks(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
workspaceID string,
|
|
issueID string,
|
|
now time.Time,
|
|
) error {
|
|
var all, notStarted, running, pending, completed, failed, canceled int
|
|
err := tx.QueryRowContext(ctx, `
|
|
SELECT COUNT(*),
|
|
SUM(CASE WHEN status = 'not_started' THEN 1 ELSE 0 END),
|
|
SUM(CASE WHEN status = 'running' THEN 1 ELSE 0 END),
|
|
SUM(CASE WHEN status = 'pending_acceptance' THEN 1 ELSE 0 END),
|
|
SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END),
|
|
SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END),
|
|
SUM(CASE WHEN status = 'canceled' THEN 1 ELSE 0 END)
|
|
FROM workspace_issue_tasks
|
|
WHERE workspace_id = ? AND issue_id = ? AND superseded_at_unix_ms = 0
|
|
`, workspaceID, issueID).Scan(&all, ¬Started, &running, &pending, &completed, &failed, &canceled)
|
|
if err != nil {
|
|
return fmt.Errorf("project scheduled Tutti mode Issue: %w", err)
|
|
}
|
|
status := workspaceissues.ProjectIssueStatus(workspaceissues.StatusCounts{
|
|
All: all, NotStarted: notStarted, Running: running,
|
|
PendingAcceptance: pending, Completed: completed, Failed: failed, Canceled: canceled,
|
|
})
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_issues
|
|
SET status = ?, task_count = ?, not_started_count = ?, running_count = ?,
|
|
pending_acceptance_count = ?, completed_count = ?, failed_count = ?,
|
|
canceled_count = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND issue_id = ?
|
|
`, string(status), all, notStarted, running, pending, completed, failed, canceled,
|
|
unixMs(now), workspaceID, issueID)
|
|
if err != nil {
|
|
return fmt.Errorf("update scheduled Tutti mode Issue projection: %w", err)
|
|
}
|
|
return requireRowsAffected(result, workspaceissues.ErrIssueNotFound, "update scheduled Tutti mode Issue projection")
|
|
}
|
|
|
|
func resolveTuttiModeScheduleCheckpoint(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
executionID string,
|
|
admission executionbiz.ScheduleAdmission,
|
|
) error {
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_tutti_execution_checkpoints
|
|
SET status = 'resolved', updated_at_unix_ms = ?, resolved_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND execution_id = ? AND checkpoint_id = ?
|
|
AND status = 'active' AND graph_revision = ?
|
|
`, unixMs(admission.Now), unixMs(admission.Now), admission.WorkspaceID,
|
|
executionID, admission.CheckpointID, admission.ExpectedGraphRevision)
|
|
if err != nil {
|
|
return fmt.Errorf("resolve Tutti mode schedule checkpoint: %w", err)
|
|
}
|
|
if err := requireRowsAffected(result, executionbiz.ErrScheduleRejected, "resolve Tutti mode schedule checkpoint"); err != nil {
|
|
return err
|
|
}
|
|
if err := acknowledgeTuttiModeCheckpointWakesTx(
|
|
ctx, tx, admission.WorkspaceID, executionID, admission.CheckpointID, admission.Now,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
nextStatus := executionbiz.StatusRunning
|
|
var nextCheckpointID string
|
|
err = tx.QueryRowContext(ctx, `
|
|
SELECT checkpoint_id
|
|
FROM workspace_tutti_execution_checkpoints
|
|
WHERE workspace_id = ? AND execution_id = ? AND status = 'pending'
|
|
ORDER BY sequence ASC
|
|
LIMIT 1
|
|
`, admission.WorkspaceID, executionID).Scan(&nextCheckpointID)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("get next Tutti mode schedule checkpoint: %w", err)
|
|
}
|
|
if err == nil {
|
|
result, err = tx.ExecContext(ctx, `
|
|
UPDATE workspace_tutti_execution_checkpoints
|
|
SET status = 'active', graph_revision = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND execution_id = ? AND checkpoint_id = ?
|
|
AND status = 'pending'
|
|
`, admission.ExpectedGraphRevision, unixMs(admission.Now),
|
|
admission.WorkspaceID, executionID, nextCheckpointID)
|
|
if err != nil {
|
|
return fmt.Errorf("promote next Tutti mode schedule checkpoint: %w", err)
|
|
}
|
|
if err := requireRowsAffected(result, executionbiz.ErrScheduleRejected, "promote next Tutti mode schedule checkpoint"); err != nil {
|
|
return err
|
|
}
|
|
nextCheckpoint, found, err := getTuttiModeCheckpointTx(
|
|
ctx, tx, admission.WorkspaceID, executionID, nextCheckpointID,
|
|
)
|
|
if err != nil || !found {
|
|
if err == nil {
|
|
err = executionbiz.ErrScheduleRejected
|
|
}
|
|
return err
|
|
}
|
|
if err := prepareTuttiModeMainWakeTx(
|
|
ctx, tx, admission.WorkspaceID, executionID, nextCheckpoint, admission.Now,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
nextStatus = executionbiz.StatusAwaitingMain
|
|
}
|
|
result, err = tx.ExecContext(ctx, `
|
|
UPDATE workspace_tutti_executions
|
|
SET status = ?, last_orchestrator_activity_at_unix_ms = ?,
|
|
watchdog_due_at_unix_ms = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND execution_id = ? AND graph_revision = ?
|
|
`, string(nextStatus), unixMs(admission.Now), unixMs(admission.Now.Add(executionbiz.WatchdogInterval)),
|
|
unixMs(admission.Now), admission.WorkspaceID, executionID,
|
|
admission.ExpectedGraphRevision)
|
|
if err != nil {
|
|
return fmt.Errorf("advance Tutti mode execution after schedule: %w", err)
|
|
}
|
|
return requireRowsAffected(result, executionbiz.ErrScheduleRejected, "advance Tutti mode execution after schedule")
|
|
}
|
|
|
|
func (s *SQLiteStore) ListPreparedTuttiModeRunLaunches(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
runIDs []string,
|
|
_ time.Time,
|
|
) ([]executionbiz.PreparedRunLaunch, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return nil, err
|
|
}
|
|
where := []string{
|
|
"runs.workspace_id = ?",
|
|
"runs.status = 'running'",
|
|
"intents.status = 'prepared'",
|
|
}
|
|
args := []any{strings.TrimSpace(workspaceID)}
|
|
if strings.TrimSpace(issueID) != "" {
|
|
where = append(where, "runs.issue_id = ?")
|
|
args = append(args, strings.TrimSpace(issueID))
|
|
}
|
|
placeholders := make([]string, len(runIDs))
|
|
for index, runID := range runIDs {
|
|
placeholders[index] = "?"
|
|
args = append(args, strings.TrimSpace(runID))
|
|
}
|
|
if len(placeholders) > 0 {
|
|
where = append(where, "runs.run_id IN ("+strings.Join(placeholders, ",")+")")
|
|
}
|
|
rows, err := s.readDB.QueryContext(ctx, fmt.Sprintf(`
|
|
SELECT %s, intents.client_submit_id
|
|
FROM workspace_issue_runs AS runs
|
|
JOIN workspace_issue_run_launch_intents AS intents
|
|
ON intents.workspace_id = runs.workspace_id
|
|
AND intents.issue_id = runs.issue_id
|
|
AND intents.task_id = runs.task_id
|
|
AND intents.run_id = runs.run_id
|
|
JOIN workspace_issues AS issues
|
|
ON issues.workspace_id = runs.workspace_id
|
|
AND issues.issue_id = runs.issue_id
|
|
JOIN workspace_tutti_executions AS executions
|
|
ON executions.workspace_id = runs.workspace_id
|
|
AND executions.issue_id = runs.issue_id
|
|
WHERE %s
|
|
AND issues.dispatch_paused = 0
|
|
AND executions.status = 'running'
|
|
ORDER BY runs.created_at_unix_ms ASC, runs.id ASC
|
|
`, prefixedRunSelectColumns("runs"), strings.Join(where, " AND ")), args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list prepared Tutti mode Run launches: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
runs := make([]executionbiz.PreparedRunLaunch, 0, len(runIDs))
|
|
for rows.Next() {
|
|
run, clientSubmitID, scanErr := scanPreparedIssueRunLaunch(rows, nil)
|
|
if scanErr != nil {
|
|
return nil, scanErr
|
|
}
|
|
runs = append(runs, executionbiz.PreparedRunLaunch{
|
|
Run: run, ClientSubmitID: clientSubmitID,
|
|
})
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate prepared Tutti mode Run launches: %w", err)
|
|
}
|
|
return runs, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) MarkTuttiModeRunLaunchIntentDispatched(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
runID string,
|
|
leaseOwner string,
|
|
now time.Time,
|
|
) error {
|
|
return s.markIssueRunLaunchIntentDispatched(
|
|
ctx, workspaceID, issueID, runID, leaseOwner, now,
|
|
executionbiz.ErrScheduleRejected,
|
|
)
|
|
}
|
|
|
|
func (s *SQLiteStore) GetTuttiModeRunLaunchClientSubmitID(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
runID string,
|
|
) (string, bool, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return "", false, err
|
|
}
|
|
var clientSubmitID string
|
|
err := s.readDB.QueryRowContext(ctx, `
|
|
SELECT client_submit_id
|
|
FROM workspace_issue_run_launch_intents
|
|
WHERE workspace_id = ? AND issue_id = ? AND run_id = ?
|
|
`, strings.TrimSpace(workspaceID), strings.TrimSpace(issueID), strings.TrimSpace(runID)).
|
|
Scan(&clientSubmitID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, nil
|
|
}
|
|
if err != nil {
|
|
return "", false, fmt.Errorf("get Tutti mode Run launch submit identity: %w", err)
|
|
}
|
|
return clientSubmitID, true, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) ClaimTuttiModeRunLaunchIntent(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
runID string,
|
|
leaseOwner string,
|
|
now time.Time,
|
|
leaseExpires time.Time,
|
|
) (bool, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return false, err
|
|
}
|
|
result, err := s.writeDB.ExecContext(ctx, `
|
|
UPDATE workspace_issue_run_launch_intents
|
|
SET status = 'leased', attempt_count = attempt_count + 1,
|
|
lease_owner = ?, lease_expires_at_unix_ms = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND issue_id = ? AND run_id = ?
|
|
AND status = 'prepared'
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM workspace_issue_runs AS runs
|
|
JOIN workspace_issues AS issues
|
|
ON issues.workspace_id = runs.workspace_id
|
|
AND issues.issue_id = runs.issue_id
|
|
JOIN workspace_tutti_executions AS executions
|
|
ON executions.workspace_id = runs.workspace_id
|
|
AND executions.issue_id = runs.issue_id
|
|
WHERE runs.workspace_id = workspace_issue_run_launch_intents.workspace_id
|
|
AND runs.issue_id = workspace_issue_run_launch_intents.issue_id
|
|
AND runs.run_id = workspace_issue_run_launch_intents.run_id
|
|
AND runs.status = 'running'
|
|
AND issues.dispatch_paused = 0
|
|
AND executions.status = 'running'
|
|
)
|
|
`, strings.TrimSpace(leaseOwner), unixMs(leaseExpires), unixMs(now),
|
|
strings.TrimSpace(workspaceID), strings.TrimSpace(issueID),
|
|
strings.TrimSpace(runID))
|
|
if err != nil {
|
|
return false, fmt.Errorf("claim Tutti mode Run launch intent: %w", err)
|
|
}
|
|
return rowsWereAffected(result, "claim Tutti mode Run launch intent")
|
|
}
|
|
|
|
func (s *SQLiteStore) ReleaseTuttiModeRunLaunchIntent(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
runID string,
|
|
leaseOwner string,
|
|
now time.Time,
|
|
) error {
|
|
return s.releaseIssueRunLaunchIntent(
|
|
ctx, workspaceID, issueID, runID, leaseOwner, now,
|
|
executionbiz.ErrScheduleRejected,
|
|
)
|
|
}
|
|
|
|
func (s *SQLiteStore) RenewTuttiModeRunLaunchIntent(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
runID string,
|
|
leaseOwner string,
|
|
now time.Time,
|
|
leaseExpires time.Time,
|
|
) error {
|
|
return s.renewIssueRunLaunchIntent(
|
|
ctx, workspaceID, issueID, runID, leaseOwner, now, leaseExpires,
|
|
executionbiz.ErrScheduleRejected,
|
|
)
|
|
}
|
|
|
|
func (s *SQLiteStore) RequeueLeasedTuttiModeRunLaunchIntents(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
now time.Time,
|
|
) error {
|
|
return s.requeueLeasedIssueRunLaunchIntents(ctx, workspaceID, now)
|
|
}
|
|
|
|
func scanPreparedIssueRunLaunch(
|
|
scanner issueScanner,
|
|
payloadJSON *string,
|
|
) (workspaceissues.Run, string, error) {
|
|
var item workspaceissues.Run
|
|
var id int64
|
|
var status string
|
|
var clientSubmitID string
|
|
destinations := []any{
|
|
&id, &item.RunID, &item.TaskID, &item.IssueID, &item.WorkspaceID,
|
|
&item.RequesterUserID, &item.AgentUserID, &item.AgentTargetID,
|
|
&item.AgentSessionID, &item.AgentProvider, &item.ModelPlanID, &item.Model,
|
|
&item.ReasoningIntensity, &status, &item.Usage.InputTokens,
|
|
&item.Usage.OutputTokens, &item.Usage.CacheReadTokens, &item.Usage.CacheWriteTokens,
|
|
&item.Summary, &item.ErrorMessage, &item.OutputDir, &item.ExecutionDirectory,
|
|
&item.CreatedAtUnixMS, &item.StartedAtUnixMS, &item.CompletedAtUnixMS,
|
|
&item.UpdatedAtUnixMS, &clientSubmitID,
|
|
}
|
|
if payloadJSON != nil {
|
|
destinations = append(destinations, payloadJSON)
|
|
}
|
|
err := scanner.Scan(destinations...)
|
|
item.ID = uint64(id)
|
|
item.Status = workspaceissues.Status(status)
|
|
return item, clientSubmitID, err
|
|
}
|
|
|
|
func prefixedRunSelectColumns(alias string) string {
|
|
columns := strings.Split(strings.TrimSpace(runSelectColumns), ",")
|
|
for index, column := range columns {
|
|
columns[index] = alias + "." + strings.TrimSpace(column)
|
|
}
|
|
return strings.Join(columns, ", ")
|
|
}
|