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
690 lines
26 KiB
Go
690 lines
26 KiB
Go
package workspace
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"strings"
|
|
"time"
|
|
|
|
executionbiz "github.com/tutti-os/tutti/services/tuttid/biz/tuttimodeexecution"
|
|
)
|
|
|
|
func (s *SQLiteStore) RequestTuttiModeArchive(
|
|
ctx context.Context,
|
|
request executionbiz.ArchiveRequest,
|
|
) (executionbiz.ArchiveOperation, bool, error) {
|
|
request.WorkspaceID = strings.TrimSpace(request.WorkspaceID)
|
|
request.IssueID = strings.TrimSpace(request.IssueID)
|
|
request.RequestID = strings.TrimSpace(request.RequestID)
|
|
request.RequestedBy = strings.TrimSpace(request.RequestedBy)
|
|
request.Reason = strings.TrimSpace(request.Reason)
|
|
request.SourceSessionID = strings.TrimSpace(request.SourceSessionID)
|
|
request.CheckpointID = strings.TrimSpace(request.CheckpointID)
|
|
request.Now = request.Now.UTC()
|
|
sourceScoped := request.SourceSessionID != "" || request.CheckpointID != "" ||
|
|
request.ExpectedGraphRevision != 0
|
|
if s == nil || s.writeDB == nil || request.WorkspaceID == "" || request.IssueID == "" ||
|
|
request.RequestID == "" || request.RequestedBy == "" || request.Reason == "" || request.Now.IsZero() {
|
|
return executionbiz.ArchiveOperation{}, false, executionbiz.ErrInvalidExecution
|
|
}
|
|
if sourceScoped && (request.SourceSessionID == "" || request.CheckpointID == "" ||
|
|
request.ExpectedGraphRevision < 1 || request.RequestedBy != request.SourceSessionID) {
|
|
return executionbiz.ArchiveOperation{}, false, executionbiz.ErrInvalidExecution
|
|
}
|
|
tx, err := s.writeDB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("begin Tutti archive request: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
execution, err := getTuttiArchiveExecutionTx(ctx, tx, request.WorkspaceID, request.IssueID)
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, err
|
|
}
|
|
operation, replayed, err := requestTuttiModeArchiveTx(
|
|
ctx, tx, request, execution, sourceScoped,
|
|
)
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("commit Tutti archive request: %w", err)
|
|
}
|
|
return operation, replayed, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) RequestTuttiModeArchivesForSourceSession(
|
|
ctx context.Context,
|
|
request executionbiz.SourceSessionArchiveRequest,
|
|
) ([]executionbiz.ArchiveOperation, error) {
|
|
request.WorkspaceID = strings.TrimSpace(request.WorkspaceID)
|
|
request.SourceSessionID = strings.TrimSpace(request.SourceSessionID)
|
|
request.RequestID = strings.TrimSpace(request.RequestID)
|
|
request.RequestedBy = strings.TrimSpace(request.RequestedBy)
|
|
request.Reason = strings.TrimSpace(request.Reason)
|
|
request.Now = request.Now.UTC()
|
|
if s == nil || s.writeDB == nil || request.WorkspaceID == "" ||
|
|
request.SourceSessionID == "" || request.RequestID == "" ||
|
|
request.RequestedBy == "" || request.Reason == "" || request.Now.IsZero() {
|
|
return nil, executionbiz.ErrInvalidExecution
|
|
}
|
|
tx, err := s.writeDB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("begin source-session Tutti archive request: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
operations, err := requestTuttiModeArchivesForSourceSessionTx(ctx, tx, request)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return nil, fmt.Errorf("commit source-session Tutti archive request: %w", err)
|
|
}
|
|
return operations, nil
|
|
}
|
|
|
|
func requestTuttiModeArchivesForSourceSessionTx(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
request executionbiz.SourceSessionArchiveRequest,
|
|
) ([]executionbiz.ArchiveOperation, error) {
|
|
rows, err := tx.QueryContext(ctx, `
|
|
SELECT execution_id, issue_id, source_session_id, status, graph_revision
|
|
FROM workspace_tutti_executions
|
|
WHERE workspace_id = ? AND source_session_id = ?
|
|
AND status NOT IN ('completed', 'archived')
|
|
ORDER BY created_at_unix_ms, execution_id
|
|
`, request.WorkspaceID, request.SourceSessionID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list source-session Tutti executions: %w", err)
|
|
}
|
|
var executions []executionbiz.Execution
|
|
for rows.Next() {
|
|
var execution executionbiz.Execution
|
|
var status string
|
|
if err := rows.Scan(
|
|
&execution.ID, &execution.IssueID, &execution.SourceSessionID,
|
|
&status, &execution.GraphRevision,
|
|
); err != nil {
|
|
_ = rows.Close()
|
|
return nil, fmt.Errorf("scan source-session Tutti execution: %w", err)
|
|
}
|
|
execution.WorkspaceID = request.WorkspaceID
|
|
execution.Status = executionbiz.Status(status)
|
|
executions = append(executions, execution)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
_ = rows.Close()
|
|
return nil, fmt.Errorf("iterate source-session Tutti executions: %w", err)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, fmt.Errorf("close source-session Tutti executions: %w", err)
|
|
}
|
|
operations := make([]executionbiz.ArchiveOperation, 0, len(executions))
|
|
for _, execution := range executions {
|
|
operation, _, err := requestTuttiModeArchiveTx(ctx, tx, executionbiz.ArchiveRequest{
|
|
WorkspaceID: request.WorkspaceID, IssueID: execution.IssueID,
|
|
RequestID: request.RequestID, RequestedBy: request.RequestedBy,
|
|
Reason: request.Reason, Now: request.Now,
|
|
}, execution, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
operations = append(operations, operation)
|
|
}
|
|
if err := cancelTuttiSourceSessionWorkflowsTx(
|
|
ctx, tx, request.WorkspaceID, request.SourceSessionID, request.Now,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
return operations, nil
|
|
}
|
|
|
|
func requestTuttiModeArchiveTx(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
request executionbiz.ArchiveRequest,
|
|
execution executionbiz.Execution,
|
|
sourceScoped bool,
|
|
) (executionbiz.ArchiveOperation, bool, error) {
|
|
operation, found, err := getTuttiArchiveOperationByRequestTx(
|
|
ctx, tx, request.WorkspaceID, execution.ID, request.RequestID,
|
|
)
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, err
|
|
}
|
|
if found {
|
|
if operation.RequestedBy != request.RequestedBy || operation.Reason != request.Reason {
|
|
return executionbiz.ArchiveOperation{}, false, executionbiz.ErrExecutionConflict
|
|
}
|
|
return operation, true, nil
|
|
}
|
|
if sourceScoped {
|
|
if execution.SourceSessionID != request.SourceSessionID ||
|
|
execution.GraphRevision != request.ExpectedGraphRevision {
|
|
return executionbiz.ArchiveOperation{}, false, executionbiz.ErrExecutionConflict
|
|
}
|
|
var activeCheckpoint int
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT COUNT(*)
|
|
FROM workspace_tutti_execution_checkpoints
|
|
WHERE workspace_id = ? AND execution_id = ? AND checkpoint_id = ?
|
|
AND status = 'active' AND graph_revision = ?
|
|
`, request.WorkspaceID, execution.ID, request.CheckpointID,
|
|
request.ExpectedGraphRevision).Scan(&activeCheckpoint); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false,
|
|
fmt.Errorf("validate source-scoped Tutti archive checkpoint: %w", err)
|
|
}
|
|
if activeCheckpoint != 1 {
|
|
return executionbiz.ArchiveOperation{}, false, executionbiz.ErrExecutionConflict
|
|
}
|
|
}
|
|
if execution.Status == executionbiz.StatusArchiving {
|
|
return currentTuttiArchiveOperationTx(ctx, tx, request.WorkspaceID, execution.ID)
|
|
}
|
|
if execution.Status == executionbiz.StatusArchived {
|
|
return latestCompletedTuttiArchiveOperationTx(ctx, tx, request.WorkspaceID, execution.ID)
|
|
}
|
|
|
|
operation = executionbiz.ArchiveOperation{
|
|
WorkspaceID: request.WorkspaceID, ExecutionID: execution.ID, IssueID: request.IssueID,
|
|
OperationID: "archive:" + execution.ID + ":" + request.RequestID,
|
|
RequestID: request.RequestID, Status: executionbiz.ArchiveStatusCancelingRuns,
|
|
RequestedBy: request.RequestedBy, Reason: request.Reason,
|
|
CreatedAt: request.Now, UpdatedAt: request.Now,
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
INSERT INTO workspace_tutti_archive_operations (
|
|
workspace_id, execution_id, operation_id, request_id, status,
|
|
requested_by, reason, created_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`, operation.WorkspaceID, operation.ExecutionID, operation.OperationID, operation.RequestID,
|
|
string(operation.Status), operation.RequestedBy, operation.Reason,
|
|
unixMs(operation.CreatedAt), unixMs(operation.UpdatedAt)); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("create Tutti archive operation: %w", err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_tutti_executions
|
|
SET status = 'archiving', updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND execution_id = ? AND status <> 'archived'
|
|
`, unixMs(request.Now), request.WorkspaceID, execution.ID); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("fence Tutti execution for archive: %w", err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_issues SET dispatch_paused = 1, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND issue_id = ?
|
|
`, unixMs(request.Now), request.WorkspaceID, request.IssueID); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("pause Tutti Issue for archive: %w", err)
|
|
}
|
|
if err := cancelTuttiArchiveOperationsTx(ctx, tx, operation, request.Now); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, err
|
|
}
|
|
return operation, false, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) GetTuttiModeArchiveOperation(
|
|
ctx context.Context, workspaceID, operationID string,
|
|
) (executionbiz.ArchiveOperation, error) {
|
|
row := s.readDB.QueryRowContext(ctx, archiveOperationSelect+`
|
|
WHERE o.workspace_id = ? AND o.operation_id = ?
|
|
`, strings.TrimSpace(workspaceID), strings.TrimSpace(operationID))
|
|
operation, err := scanTuttiArchiveOperation(row)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return executionbiz.ArchiveOperation{}, executionbiz.ErrExecutionNotFound
|
|
}
|
|
return operation, err
|
|
}
|
|
|
|
func (s *SQLiteStore) FailTuttiModeArchive(
|
|
ctx context.Context, workspaceID, operationID, message string, now time.Time,
|
|
) (executionbiz.ArchiveOperation, error) {
|
|
_, err := s.writeDB.ExecContext(ctx, `
|
|
UPDATE workspace_tutti_archive_operations
|
|
SET status = 'failed', attempt_count = attempt_count + 1, last_error = ?,
|
|
updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND operation_id = ? AND status <> 'completed'
|
|
`, strings.TrimSpace(message), unixMs(now.UTC()), strings.TrimSpace(workspaceID), strings.TrimSpace(operationID))
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, fmt.Errorf("fail Tutti archive operation: %w", err)
|
|
}
|
|
return s.GetTuttiModeArchiveOperation(ctx, workspaceID, operationID)
|
|
}
|
|
|
|
func (s *SQLiteStore) CompleteTuttiModeArchiveIfSettled(
|
|
ctx context.Context, workspaceID, operationID string, now time.Time,
|
|
) (executionbiz.ArchiveOperation, bool, error) {
|
|
tx, err := s.writeDB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("begin Tutti archive completion: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
operation, err := scanTuttiArchiveOperation(tx.QueryRowContext(ctx, archiveOperationSelect+`
|
|
WHERE o.workspace_id = ? AND o.operation_id = ?
|
|
`, strings.TrimSpace(workspaceID), strings.TrimSpace(operationID)))
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, err
|
|
}
|
|
if operation.Status == executionbiz.ArchiveStatusCompleted {
|
|
return operation, true, nil
|
|
}
|
|
var running int
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM workspace_issue_runs
|
|
WHERE workspace_id = ? AND issue_id = ? AND status = 'running'
|
|
`, operation.WorkspaceID, operation.IssueID).Scan(&running); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("count archive running Runs: %w", err)
|
|
}
|
|
if running > 0 {
|
|
_, err = tx.ExecContext(ctx, `
|
|
UPDATE workspace_tutti_archive_operations
|
|
SET status = 'archiving', last_error = '', updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND operation_id = ?
|
|
`, unixMs(now.UTC()), operation.WorkspaceID, operation.OperationID)
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, err
|
|
}
|
|
operation.Status = executionbiz.ArchiveStatusArchiving
|
|
operation.LastError = ""
|
|
operation.UpdatedAt = now.UTC()
|
|
return operation, false, nil
|
|
}
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_tutti_executions
|
|
SET status = 'archived', archived_at_unix_ms = ?, archived_by = ?,
|
|
archive_reason = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND execution_id = ? AND status = 'archiving'
|
|
`, unixMs(now.UTC()), operation.RequestedBy, operation.Reason, unixMs(now.UTC()),
|
|
operation.WorkspaceID, operation.ExecutionID)
|
|
if err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("seal archived Tutti execution: %w", err)
|
|
}
|
|
if err := requireRowsAffected(
|
|
result,
|
|
executionbiz.ErrExecutionConflict,
|
|
"seal archived Tutti execution",
|
|
); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_tutti_archive_operations
|
|
SET status = 'completed', last_error = '', updated_at_unix_ms = ?, completed_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND operation_id = ?
|
|
`, unixMs(now.UTC()), unixMs(now.UTC()), operation.WorkspaceID, operation.OperationID); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("complete Tutti archive operation: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return executionbiz.ArchiveOperation{}, false, fmt.Errorf("commit Tutti archive completion: %w", err)
|
|
}
|
|
operation.Status = executionbiz.ArchiveStatusCompleted
|
|
operation.LastError = ""
|
|
operation.UpdatedAt = now.UTC()
|
|
operation.CompletedAt = now.UTC()
|
|
return operation, true, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) ListRecoverableTuttiModeArchives(
|
|
ctx context.Context, workspaceID string,
|
|
) ([]executionbiz.ArchiveOperation, error) {
|
|
rows, err := s.readDB.QueryContext(ctx, archiveOperationSelect+`
|
|
WHERE o.workspace_id = ? AND o.status IN ('requested', 'canceling_runs', 'archiving', 'failed')
|
|
ORDER BY o.created_at_unix_ms, o.operation_id
|
|
`, strings.TrimSpace(workspaceID))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []executionbiz.ArchiveOperation
|
|
for rows.Next() {
|
|
operation, scanErr := scanTuttiArchiveOperation(rows)
|
|
if scanErr != nil {
|
|
return nil, scanErr
|
|
}
|
|
result = append(result, operation)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *SQLiteStore) AdmitSourceSessionDeletion(
|
|
ctx context.Context, input executionbiz.SourceSessionDeletionAdmission,
|
|
) (executionbiz.SourceSessionDeletionAdmission, error) {
|
|
input.WorkspaceID = strings.TrimSpace(input.WorkspaceID)
|
|
input.SessionIDs = normalizedSourceSessionIDs(input.SessionIDs)
|
|
slices.Sort(input.SessionIDs)
|
|
input.Now = input.Now.UTC()
|
|
if input.WorkspaceID == "" || len(input.SessionIDs) == 0 || input.Now.IsZero() {
|
|
return executionbiz.SourceSessionDeletionAdmission{}, errors.New("source session deletion admission is incomplete")
|
|
}
|
|
closureJSON, _ := json.Marshal(input.SessionIDs)
|
|
sum := sha256.Sum256(closureJSON)
|
|
input.AdmissionID = "source-delete:" + hex.EncodeToString(sum[:])
|
|
tx, err := s.writeDB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return executionbiz.SourceSessionDeletionAdmission{}, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
protected, err := listProtectedTuttiSourcesTx(ctx, tx, input.WorkspaceID, input.SessionIDs)
|
|
if err != nil {
|
|
return executionbiz.SourceSessionDeletionAdmission{}, err
|
|
}
|
|
if len(protected) > 0 {
|
|
return executionbiz.SourceSessionDeletionAdmission{}, &executionbiz.ProtectedSourceError{
|
|
WorkspaceID: input.WorkspaceID, Issues: protected,
|
|
}
|
|
}
|
|
_, err = tx.ExecContext(ctx, `
|
|
INSERT INTO workspace_source_session_deletion_admissions (
|
|
workspace_id, admission_id, closure_sha256, closure_json, status,
|
|
created_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, 'admitted', ?, ?)
|
|
ON CONFLICT(workspace_id, closure_sha256) DO UPDATE SET
|
|
admission_id = excluded.admission_id, closure_json = excluded.closure_json,
|
|
status = 'admitted', updated_at_unix_ms = excluded.updated_at_unix_ms,
|
|
finalized_at_unix_ms = 0
|
|
`, input.WorkspaceID, input.AdmissionID, hex.EncodeToString(sum[:]), string(closureJSON),
|
|
unixMs(input.Now), unixMs(input.Now))
|
|
if err != nil {
|
|
return executionbiz.SourceSessionDeletionAdmission{}, fmt.Errorf("record source deletion admission: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return executionbiz.SourceSessionDeletionAdmission{}, err
|
|
}
|
|
return input, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) ReportSourceSessionDeletion(
|
|
ctx context.Context, input executionbiz.SourceSessionDeletionAdmission, succeeded bool, now time.Time,
|
|
) error {
|
|
if strings.TrimSpace(input.AdmissionID) == "" {
|
|
sessionIDs := normalizedSourceSessionIDs(input.SessionIDs)
|
|
slices.Sort(sessionIDs)
|
|
closureJSON, _ := json.Marshal(sessionIDs)
|
|
sum := sha256.Sum256(closureJSON)
|
|
input.AdmissionID = "source-delete:" + hex.EncodeToString(sum[:])
|
|
}
|
|
status := "superseded"
|
|
finalized := int64(0)
|
|
if succeeded {
|
|
status = "finalized"
|
|
finalized = unixMs(now.UTC())
|
|
}
|
|
_, err := s.writeDB.ExecContext(ctx, `
|
|
UPDATE workspace_source_session_deletion_admissions
|
|
SET status = ?, updated_at_unix_ms = ?, finalized_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND admission_id = ?
|
|
`, status, unixMs(now.UTC()), finalized, strings.TrimSpace(input.WorkspaceID), strings.TrimSpace(input.AdmissionID))
|
|
return err
|
|
}
|
|
|
|
func (s *SQLiteStore) ReconcileSourceSessionDeletionAdmissions(
|
|
ctx context.Context, now time.Time,
|
|
) error {
|
|
if s == nil || s.writeDB == nil || now.IsZero() {
|
|
return errors.New("source session deletion admission recovery is incomplete")
|
|
}
|
|
_, err := s.writeDB.ExecContext(ctx, `
|
|
UPDATE workspace_source_session_deletion_admissions
|
|
SET status = 'superseded', updated_at_unix_ms = ?
|
|
WHERE status = 'admitted'
|
|
`, unixMs(now.UTC()))
|
|
if err != nil {
|
|
return fmt.Errorf("release abandoned source deletion admissions: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func ensureSourceSessionNotDeletionFencedTx(
|
|
ctx context.Context, tx *sql.Tx, workspaceID, sourceSessionID string,
|
|
) error {
|
|
var count int
|
|
err := tx.QueryRowContext(ctx, `
|
|
SELECT COUNT(*)
|
|
FROM workspace_source_session_deletion_admissions a, json_each(a.closure_json) c
|
|
WHERE a.workspace_id = ? AND a.status = 'admitted' AND c.value = ?
|
|
`, workspaceID, sourceSessionID).Scan(&count)
|
|
if err != nil {
|
|
return fmt.Errorf("check source deletion materialization fence: %w", err)
|
|
}
|
|
if count > 0 {
|
|
return executionbiz.ErrSourceDeletionFenced
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func cancelTuttiArchiveOperationsTx(
|
|
ctx context.Context, tx *sql.Tx, operation executionbiz.ArchiveOperation, now time.Time,
|
|
) error {
|
|
updates := []struct {
|
|
statement string
|
|
ownerID string
|
|
}{
|
|
{statement: `UPDATE workspace_tutti_execution_checkpoints SET status = 'canceled', updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND execution_id = ? AND status IN ('pending', 'active')`,
|
|
ownerID: operation.ExecutionID},
|
|
{statement: `UPDATE workspace_tutti_execution_wakes SET status = 'canceled', lease_owner = '',
|
|
lease_expires_at_unix_ms = 0, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND execution_id = ?
|
|
AND status IN ('prepared', 'leased', 'dispatched', 'turn_settled')`,
|
|
ownerID: operation.ExecutionID},
|
|
{statement: `UPDATE workspace_tutti_goal_reviews SET status = 'canceled', updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND execution_id = ? AND status IN ('prepared', 'leased', 'dispatched')`,
|
|
ownerID: operation.ExecutionID},
|
|
{statement: `UPDATE workspace_issue_run_launch_intents SET status = 'canceled', lease_owner = '',
|
|
lease_expires_at_unix_ms = 0, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND issue_id = ? AND status IN ('prepared', 'leased')`,
|
|
ownerID: operation.IssueID},
|
|
}
|
|
for _, update := range updates {
|
|
if _, err := tx.ExecContext(
|
|
ctx, update.statement, unixMs(now), operation.WorkspaceID, update.ownerID,
|
|
); err != nil {
|
|
return fmt.Errorf("fence Tutti archive child operation: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func cancelTuttiSourceSessionWorkflowsTx(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
workspaceID string,
|
|
sourceSessionID string,
|
|
now time.Time,
|
|
) error {
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_workflow_checkpoints
|
|
SET status = 'canceled', updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND status = 'pending'
|
|
AND workflow_id IN (
|
|
SELECT workflow_id FROM workspace_workflows
|
|
WHERE workspace_id = ? AND source_session_id = ?
|
|
AND status IN ('pending_review', 'in_progress', 'accepted')
|
|
)
|
|
`, unixMs(now), workspaceID, workspaceID, sourceSessionID); err != nil {
|
|
return fmt.Errorf("cancel source-session workflow checkpoints: %w", err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_workflows
|
|
SET status = 'canceled', updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND source_session_id = ?
|
|
AND (
|
|
status IN ('pending_review', 'in_progress')
|
|
OR (
|
|
status = 'accepted'
|
|
AND (
|
|
EXISTS (
|
|
SELECT 1 FROM workspace_tutti_executions execution
|
|
WHERE execution.workspace_id = workspace_workflows.workspace_id
|
|
AND execution.workflow_id = workspace_workflows.workflow_id
|
|
AND execution.status NOT IN ('completed', 'archived')
|
|
)
|
|
OR EXISTS (
|
|
SELECT 1 FROM workspace_workflow_operations operation
|
|
WHERE operation.workspace_id = workspace_workflows.workspace_id
|
|
AND operation.workflow_id = workspace_workflows.workflow_id
|
|
AND operation.status IN ('pending', 'running', 'failed')
|
|
)
|
|
)
|
|
)
|
|
)
|
|
`, unixMs(now), workspaceID, sourceSessionID); err != nil {
|
|
return fmt.Errorf("cancel source-session workflows: %w", err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_workflow_operations
|
|
SET status = 'canceled', updated_at_unix_ms = ?, completed_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND status IN ('pending', 'running', 'failed')
|
|
AND workflow_id IN (
|
|
SELECT workflow_id FROM workspace_workflows
|
|
WHERE workspace_id = ? AND source_session_id = ? AND status = 'canceled'
|
|
)
|
|
`, unixMs(now), unixMs(now), workspaceID, workspaceID, sourceSessionID); err != nil {
|
|
return fmt.Errorf("cancel source-session workflow operations: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func getTuttiArchiveExecutionTx(
|
|
ctx context.Context, tx *sql.Tx, workspaceID, issueID string,
|
|
) (executionbiz.Execution, error) {
|
|
var execution executionbiz.Execution
|
|
var status string
|
|
err := tx.QueryRowContext(ctx, `
|
|
SELECT execution_id, source_session_id, status, graph_revision
|
|
FROM workspace_tutti_executions WHERE workspace_id = ? AND issue_id = ?
|
|
`, workspaceID, issueID).Scan(
|
|
&execution.ID, &execution.SourceSessionID, &status, &execution.GraphRevision,
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return execution, executionbiz.ErrExecutionNotFound
|
|
}
|
|
execution.WorkspaceID, execution.IssueID, execution.Status = workspaceID, issueID, executionbiz.Status(status)
|
|
return execution, err
|
|
}
|
|
|
|
func getTuttiArchiveOperationByRequestTx(
|
|
ctx context.Context, tx *sql.Tx, workspaceID, executionID, requestID string,
|
|
) (executionbiz.ArchiveOperation, bool, error) {
|
|
operation, err := scanTuttiArchiveOperation(tx.QueryRowContext(ctx, archiveOperationSelect+`
|
|
WHERE o.workspace_id = ? AND o.execution_id = ? AND o.request_id = ?
|
|
`, workspaceID, executionID, requestID))
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return executionbiz.ArchiveOperation{}, false, nil
|
|
}
|
|
return operation, err == nil, err
|
|
}
|
|
|
|
func latestCompletedTuttiArchiveOperationTx(
|
|
ctx context.Context, tx *sql.Tx, workspaceID, executionID string,
|
|
) (executionbiz.ArchiveOperation, bool, error) {
|
|
operation, err := scanTuttiArchiveOperation(tx.QueryRowContext(ctx, archiveOperationSelect+`
|
|
WHERE o.workspace_id = ? AND o.execution_id = ? AND o.status = 'completed'
|
|
ORDER BY o.completed_at_unix_ms DESC LIMIT 1
|
|
`, workspaceID, executionID))
|
|
return operation, true, err
|
|
}
|
|
|
|
func currentTuttiArchiveOperationTx(
|
|
ctx context.Context, tx *sql.Tx, workspaceID, executionID string,
|
|
) (executionbiz.ArchiveOperation, bool, error) {
|
|
operation, err := scanTuttiArchiveOperation(tx.QueryRowContext(ctx, archiveOperationSelect+`
|
|
WHERE o.workspace_id = ? AND o.execution_id = ?
|
|
AND o.status IN ('requested', 'canceling_runs', 'archiving', 'failed')
|
|
ORDER BY o.created_at_unix_ms, o.operation_id LIMIT 1
|
|
`, workspaceID, executionID))
|
|
return operation, true, err
|
|
}
|
|
|
|
const archiveOperationSelect = `
|
|
SELECT o.workspace_id, o.execution_id, e.issue_id, o.operation_id, o.request_id,
|
|
o.status, o.requested_by, o.reason, o.attempt_count, o.last_error,
|
|
o.created_at_unix_ms, o.updated_at_unix_ms, o.completed_at_unix_ms
|
|
FROM workspace_tutti_archive_operations o
|
|
JOIN workspace_tutti_executions e
|
|
ON e.workspace_id = o.workspace_id AND e.execution_id = o.execution_id
|
|
`
|
|
|
|
type archiveOperationScanner interface{ Scan(...any) error }
|
|
|
|
func scanTuttiArchiveOperation(scanner archiveOperationScanner) (executionbiz.ArchiveOperation, error) {
|
|
var operation executionbiz.ArchiveOperation
|
|
var status string
|
|
var created, updated, completed int64
|
|
err := scanner.Scan(
|
|
&operation.WorkspaceID, &operation.ExecutionID, &operation.IssueID,
|
|
&operation.OperationID, &operation.RequestID, &status, &operation.RequestedBy,
|
|
&operation.Reason, &operation.AttemptCount, &operation.LastError,
|
|
&created, &updated, &completed,
|
|
)
|
|
operation.Status = executionbiz.ArchiveStatus(status)
|
|
operation.CreatedAt = optionalTuttiModeExecutionTime(created)
|
|
operation.UpdatedAt = optionalTuttiModeExecutionTime(updated)
|
|
operation.CompletedAt = optionalTuttiModeExecutionTime(completed)
|
|
return operation, err
|
|
}
|
|
|
|
func listProtectedTuttiSourcesTx(
|
|
ctx context.Context, tx *sql.Tx, workspaceID string, sessionIDs []string,
|
|
) ([]executionbiz.ProtectedIssue, error) {
|
|
args := []any{workspaceID}
|
|
args = appendSourceSessionStrings(args, sessionIDs)
|
|
rows, err := tx.QueryContext(ctx, `
|
|
SELECT issue_id, execution_id, source_session_id, status
|
|
FROM workspace_tutti_executions
|
|
WHERE workspace_id = ? AND source_session_id IN (`+sourceSessionSQLPlaceholders(len(sessionIDs))+`)
|
|
AND status NOT IN ('completed', 'archived')
|
|
ORDER BY issue_id
|
|
`, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []executionbiz.ProtectedIssue
|
|
for rows.Next() {
|
|
var item executionbiz.ProtectedIssue
|
|
var status string
|
|
if err := rows.Scan(&item.IssueID, &item.ExecutionID, &item.SourceSessionID, &status); err != nil {
|
|
return nil, err
|
|
}
|
|
item.Status = executionbiz.Status(status)
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func normalizedSourceSessionIDs(values []string) []string {
|
|
result := make([]string, 0, len(values))
|
|
seen := make(map[string]struct{}, len(values))
|
|
for _, value := range values {
|
|
value = strings.TrimSpace(value)
|
|
if value == "" {
|
|
continue
|
|
}
|
|
if _, exists := seen[value]; exists {
|
|
continue
|
|
}
|
|
seen[value] = struct{}{}
|
|
result = append(result, value)
|
|
}
|
|
return result
|
|
}
|
|
|
|
func sourceSessionSQLPlaceholders(count int) string {
|
|
return strings.TrimSuffix(strings.Repeat("?,", count), ",")
|
|
}
|
|
|
|
func appendSourceSessionStrings(args []any, values []string) []any {
|
|
for _, value := range values {
|
|
args = append(args, value)
|
|
}
|
|
return args
|
|
}
|