tutti/services/tuttid/data/workspace/sqlite_issue_run_launch.go

package workspace

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	workspaceissues "github.com/tutti-os/tutti/packages/workspace/issues"
)

// CreateIssueRunWithLaunchIntent atomically claims the task, creates the Run,
// and records the stable external-delivery identity. Recovery can therefore
// redeliver after any crash without creating a second Agent turn.
func (s *SQLiteStore) CreateIssueRunWithLaunchIntent(
	ctx context.Context,
	prepared workspaceissues.PreparedRun,
	clientSubmitID string,
	payloadJSON string,
) (workspaceissues.Run, error) {
	if err := s.ensureIssueDatabase(); err != nil {
		return workspaceissues.Run{}, err
	}
	clientSubmitID = strings.TrimSpace(clientSubmitID)
	if clientSubmitID == "" || prepared.Run.TaskID == "" ||
		prepared.Task.TaskID != prepared.Run.TaskID ||
		prepared.Issue.IssueID != prepared.Run.IssueID {
		return workspaceissues.Run{}, workspaceissues.ErrInvalidArgument
	}
	payloadJSON = strings.TrimSpace(payloadJSON)
	if !json.Valid([]byte(payloadJSON)) {
		return workspaceissues.Run{}, workspaceissues.ErrInvalidArgument
	}
	tx, err := s.writeDB.BeginTx(ctx, nil)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("begin Issue Run launch admission: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	if prepared.TaskIsNew {
		if _, err := insertWorkspaceIssueTask(ctx, tx, prepared.Task); err != nil {
			return workspaceissues.Run{}, err
		}
	}
	if err := insertWorkspaceIssueRun(ctx, tx, prepared.Run); err != nil {
		return workspaceissues.Run{}, 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 = ?
`, prepared.Run.RunID, prepared.Run.UpdatedAtUnixMS, prepared.Run.WorkspaceID,
		prepared.Run.IssueID, prepared.Run.TaskID)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("claim Issue Run launch task: %w", err)
	}
	if err := requireRowsAffected(result, workspaceissues.ErrInvalidArgument, "claim Issue Run launch task"); err != nil {
		return workspaceissues.Run{}, err
	}
	if err := projectWorkspaceIssueTasks(
		ctx, tx, prepared.Run.WorkspaceID, prepared.Run.IssueID,
		time.UnixMilli(prepared.Run.UpdatedAtUnixMS).UTC(),
	); err != nil {
		return workspaceissues.Run{}, err
	}
	result, err = tx.ExecContext(ctx, `
UPDATE workspace_issue_topics
SET last_activity_at_unix_ms = ?
WHERE workspace_id = ? AND topic_id = ?
`, prepared.Run.UpdatedAtUnixMS, prepared.Run.WorkspaceID, prepared.Issue.TopicID)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("touch topic during Issue Run launch admission: %w", err)
	}
	if err := requireRowsAffected(result, workspaceissues.ErrTopicNotFound, "touch topic during Issue Run launch admission"); err != nil {
		return workspaceissues.Run{}, 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, payload_json, created_at_unix_ms, updated_at_unix_ms
) VALUES (?, ?, ?, ?, ?, ?, 'prepared', ?, ?, ?)
`, prepared.Run.WorkspaceID, prepared.Run.IssueID, prepared.Run.TaskID,
		prepared.Run.RunID, "launch-intent:"+prepared.Run.RunID, clientSubmitID,
		payloadJSON, prepared.Run.CreatedAtUnixMS, prepared.Run.UpdatedAtUnixMS)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("create Issue Run launch intent: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return workspaceissues.Run{}, fmt.Errorf("commit Issue Run launch admission: %w", err)
	}
	return s.GetRun(ctx, prepared.Run.WorkspaceID, prepared.Run.IssueID, prepared.Run.TaskID, prepared.Run.RunID)
}

func (s *SQLiteStore) ListPreparedIssueRunLaunches(
	ctx context.Context,
	workspaceID string,
) ([]workspaceissues.PreparedRunLaunch, error) {
	if err := s.ensureIssueDatabase(); err != nil {
		return nil, err
	}
	rows, err := s.readDB.QueryContext(ctx, fmt.Sprintf(`
SELECT %s, intents.client_submit_id, intents.payload_json
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
WHERE runs.workspace_id = ? AND runs.status = 'running'
  AND intents.status = 'prepared'
  AND issues.dispatch_paused = 0
  AND issues.planning_source <> ?
ORDER BY runs.created_at_unix_ms ASC, runs.id ASC
`, prefixedRunSelectColumns("runs")), strings.TrimSpace(workspaceID),
		string(workspaceissues.PlanningSourceTuttiModePlan))
	if err != nil {
		return nil, fmt.Errorf("list prepared Issue Run launches: %w", err)
	}
	defer rows.Close()
	launches := make([]workspaceissues.PreparedRunLaunch, 0)
	for rows.Next() {
		var payloadJSON string
		run, clientSubmitID, err := scanPreparedIssueRunLaunch(rows, &payloadJSON)
		if err != nil {
			return nil, err
		}
		launches = append(launches, workspaceissues.PreparedRunLaunch{
			Run: run, ClientSubmitID: clientSubmitID, OpaquePayload: payloadJSON,
		})
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate prepared Issue Run launches: %w", err)
	}
	return launches, nil
}

// HasPendingIssueRunLaunch reports whether deleting the Issue (taskID empty)
// or one task would destroy an explicit launch while external delivery is
// still pending or in flight.
func (s *SQLiteStore) HasPendingIssueRunLaunch(
	ctx context.Context,
	workspaceID, issueID, taskID string,
) (bool, error) {
	if err := s.ensureIssueDatabase(); err != nil {
		return false, err
	}
	var found int
	err := s.readDB.QueryRowContext(ctx, `
SELECT 1
FROM workspace_issue_run_launch_intents
WHERE workspace_id = ? AND issue_id = ?
  AND status IN ('prepared', 'leased')
  AND (? = '' OR task_id = ?)
LIMIT 1
`, strings.TrimSpace(workspaceID), strings.TrimSpace(issueID), strings.TrimSpace(taskID), strings.TrimSpace(taskID)).Scan(&found)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("find pending Issue Run launch: %w", err)
	}
	return found == 1, nil
}

func (s *SQLiteStore) ClaimIssueRunLaunchIntent(
	ctx context.Context,
	workspaceID, issueID, runID, leaseOwner string,
	now, 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
    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 issues.planning_source <> ?
  )
`, strings.TrimSpace(leaseOwner), unixMs(leaseExpires), unixMs(now),
		strings.TrimSpace(workspaceID), strings.TrimSpace(issueID), strings.TrimSpace(runID),
		string(workspaceissues.PlanningSourceTuttiModePlan))
	if err != nil {
		return false, fmt.Errorf("claim Issue Run launch intent: %w", err)
	}
	return rowsWereAffected(result, "claim Issue Run launch intent")
}

func (s *SQLiteStore) RenewIssueRunLaunchIntent(ctx context.Context, workspaceID, issueID, runID, leaseOwner string, now, leaseExpires time.Time) error {
	return s.renewIssueRunLaunchIntent(
		ctx, workspaceID, issueID, runID, leaseOwner, now, leaseExpires,
		workspaceissues.ErrInvalidArgument,
	)
}

func (s *SQLiteStore) ReleaseIssueRunLaunchIntent(ctx context.Context, workspaceID, issueID, runID, leaseOwner string, now time.Time) error {
	return s.releaseIssueRunLaunchIntent(
		ctx, workspaceID, issueID, runID, leaseOwner, now,
		workspaceissues.ErrInvalidArgument,
	)
}

func (s *SQLiteStore) MarkIssueRunLaunchIntentDispatched(ctx context.Context, workspaceID, issueID, runID, leaseOwner string, now time.Time) error {
	return s.markIssueRunLaunchIntentDispatched(
		ctx, workspaceID, issueID, runID, leaseOwner, now,
		workspaceissues.ErrInvalidArgument,
	)
}

// SettleIssueRunLaunch atomically closes an authoritative pre-delivery
// failure/cancellation across the intent, Run, task projection, Issue
// projection, and topic activity.
func (s *SQLiteStore) SettleIssueRunLaunch(
	ctx context.Context,
	settlement workspaceissues.RunLaunchSettlement,
) (workspaceissues.Run, error) {
	if err := s.ensureIssueDatabase(); err != nil {
		return workspaceissues.Run{}, err
	}
	if settlement.Status != workspaceissues.StatusFailed && settlement.Status != workspaceissues.StatusCanceled {
		return workspaceissues.Run{}, workspaceissues.ErrInvalidArgument
	}
	tx, err := s.writeDB.BeginTx(ctx, nil)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("begin Issue Run launch settlement: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	var taskID string
	err = tx.QueryRowContext(ctx, `
SELECT runs.task_id
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
WHERE runs.workspace_id = ? AND runs.issue_id = ? AND runs.run_id = ?
  AND runs.status = 'running' AND issues.planning_source <> ?
`, settlement.WorkspaceID, settlement.IssueID, settlement.RunID,
		string(workspaceissues.PlanningSourceTuttiModePlan)).Scan(&taskID)
	if errors.Is(err, sql.ErrNoRows) {
		return workspaceissues.Run{}, workspaceissues.ErrRunNotFound
	}
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("get running Issue Run launch settlement: %w", err)
	}
	intentStatus := "failed"
	if settlement.Status == workspaceissues.StatusCanceled {
		intentStatus = "canceled"
	}
	result, err := tx.ExecContext(ctx, `
UPDATE workspace_issue_run_launch_intents
SET status = ?, lease_owner = '', lease_expires_at_unix_ms = 0,
    last_error = ?, updated_at_unix_ms = ?
WHERE workspace_id = ? AND issue_id = ? AND run_id = ?
  AND status = 'leased' AND lease_owner = ?
`, intentStatus, strings.TrimSpace(settlement.ErrorMessage), settlement.NowUnixMS,
		settlement.WorkspaceID, settlement.IssueID, settlement.RunID, settlement.LeaseOwner)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("settle Issue Run launch intent: %w", err)
	}
	if err := requireRowsAffected(result, workspaceissues.ErrInvalidArgument, "settle owned Issue Run launch intent"); err != nil {
		return workspaceissues.Run{}, err
	}
	result, err = tx.ExecContext(ctx, `
UPDATE workspace_issue_runs
SET status = ?, error_message = ?, completed_at_unix_ms = ?, updated_at_unix_ms = ?
WHERE workspace_id = ? AND issue_id = ? AND task_id = ? AND run_id = ?
  AND status = 'running'
`, string(settlement.Status), strings.TrimSpace(settlement.ErrorMessage), settlement.NowUnixMS,
		settlement.NowUnixMS, settlement.WorkspaceID, settlement.IssueID, taskID, settlement.RunID)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("settle Issue Run before launch: %w", err)
	}
	if err := requireRowsAffected(result, workspaceissues.ErrRunNotFound, "settle running Issue Run before launch"); err != nil {
		return workspaceissues.Run{}, err
	}
	result, err = tx.ExecContext(ctx, `
UPDATE workspace_issue_tasks
SET status = ?, acceptance_state = 'agent_claimed', acceptance_summary = '',
    updated_at_unix_ms = ?
WHERE workspace_id = ? AND issue_id = ? AND task_id = ?
  AND status = 'running' AND latest_run_id = ?
`, string(settlement.Status), settlement.NowUnixMS, settlement.WorkspaceID,
		settlement.IssueID, taskID, settlement.RunID)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("settle Issue task before launch: %w", err)
	}
	if err := requireRowsAffected(result, workspaceissues.ErrTaskNotFound, "settle running Issue task before launch"); err != nil {
		return workspaceissues.Run{}, err
	}
	if err := projectWorkspaceIssueTasks(
		ctx, tx, settlement.WorkspaceID, settlement.IssueID,
		time.UnixMilli(settlement.NowUnixMS).UTC(),
	); err != nil {
		return workspaceissues.Run{}, err
	}
	result, err = tx.ExecContext(ctx, `
UPDATE workspace_issue_topics
SET last_activity_at_unix_ms = ?
WHERE workspace_id = ? AND topic_id = (
  SELECT topic_id FROM workspace_issues WHERE workspace_id = ? AND issue_id = ?
)
`, settlement.NowUnixMS, settlement.WorkspaceID, settlement.WorkspaceID, settlement.IssueID)
	if err != nil {
		return workspaceissues.Run{}, fmt.Errorf("touch topic after Issue Run launch settlement: %w", err)
	}
	if err := requireRowsAffected(result, workspaceissues.ErrTopicNotFound, "touch topic after Issue Run launch settlement"); err != nil {
		return workspaceissues.Run{}, err
	}
	if err := tx.Commit(); err != nil {
		return workspaceissues.Run{}, fmt.Errorf("commit Issue Run launch settlement: %w", err)
	}
	return s.GetRun(ctx, settlement.WorkspaceID, settlement.IssueID, taskID, settlement.RunID)
}

func (s *SQLiteStore) RequeueLeasedIssueRunLaunchIntents(ctx context.Context, workspaceID string, now time.Time) error {
	return s.requeueLeasedIssueRunLaunchIntents(ctx, workspaceID, now)
}