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
830 lines
30 KiB
Go
830 lines
30 KiB
Go
package workspace
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
workspaceissues "github.com/tutti-os/tutti/packages/workspace/issues"
|
|
workspacebiz "github.com/tutti-os/tutti/services/tuttid/biz/workspace"
|
|
)
|
|
|
|
var _ workspaceissues.Store = (*SQLiteStore)(nil)
|
|
|
|
const issueSelectColumns = `
|
|
id, issue_id, topic_id, workspace_id, title, content, search_text, status,
|
|
planning_source, source_session_id, sequential_execution, parallel_execution, dispatch_paused, reasoning_intensity, orchestration_intensity,
|
|
budget_mode, budget_token_limit, budget_consumed_tokens,
|
|
budget_quota_waterline_percent, budget_remaining_quota_percent,
|
|
budget_has_remaining_quota, budget_status,
|
|
task_count, not_started_count, running_count, pending_acceptance_count,
|
|
completed_count, failed_count, canceled_count, creator_user_id,
|
|
creator_display_name, creator_avatar_url, created_at_unix_ms, updated_at_unix_ms`
|
|
|
|
const topicSelectColumns = `
|
|
id, topic_id, workspace_id, title, summary, is_default, pinned_at_unix_ms,
|
|
last_activity_at_unix_ms, created_at_unix_ms, updated_at_unix_ms`
|
|
|
|
const taskSelectColumns = `
|
|
id, task_id, issue_id, workspace_id, title, content, search_text, status,
|
|
priority, sort_index, due_at_unix_ms, agent_target_id, model_plan_id, model,
|
|
permission_mode_id, reasoning_effort,
|
|
execution_directory, dependency_task_ids_json, parallelizable, auto_accept,
|
|
acceptance_state, acceptance_summary, creator_user_id, creator_display_name,
|
|
creator_avatar_url, latest_run_id, superseded_at_unix_ms, superseded_by_task_id,
|
|
created_at_unix_ms, updated_at_unix_ms`
|
|
|
|
// workspaceIssueExecer abstracts the write handle (direct connection or
|
|
// in-flight transaction) used by issue and task insert helpers.
|
|
type workspaceIssueExecer interface {
|
|
ExecContext(context.Context, string, ...any) (sql.Result, error)
|
|
}
|
|
|
|
const runSelectColumns = `
|
|
id, 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`
|
|
|
|
const runOutputSelectColumns = `
|
|
id, output_id, run_id, task_id, issue_id, workspace_id, path, display_name,
|
|
media_type, size_bytes, created_at_unix_ms`
|
|
|
|
func (s *SQLiteStore) ListTopics(ctx context.Context, workspaceID string) (workspaceissues.TopicList, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.TopicList{}, err
|
|
}
|
|
if err := s.ensureIssueWorkspace(ctx, workspaceID); err != nil {
|
|
return workspaceissues.TopicList{}, err
|
|
}
|
|
|
|
rows, err := s.readDB.QueryContext(ctx, fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issue_topics
|
|
WHERE workspace_id = ?
|
|
ORDER BY
|
|
CASE WHEN pinned_at_unix_ms > 0 THEN 0 ELSE 1 END,
|
|
pinned_at_unix_ms DESC,
|
|
last_activity_at_unix_ms DESC,
|
|
id DESC
|
|
`, topicSelectColumns), workspaceID)
|
|
if err != nil {
|
|
return workspaceissues.TopicList{}, fmt.Errorf("list workspace issue topics: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
items, err := scanWorkspaceIssueTopics(rows)
|
|
if err != nil {
|
|
return workspaceissues.TopicList{}, err
|
|
}
|
|
return workspaceissues.TopicList{Items: items}, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) CreateTopic(ctx context.Context, topic workspaceissues.Topic) (workspaceissues.Topic, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Topic{}, err
|
|
}
|
|
if err := s.ensureIssueWorkspace(ctx, topic.WorkspaceID); err != nil {
|
|
return workspaceissues.Topic{}, err
|
|
}
|
|
|
|
_, err := s.writeDB.ExecContext(ctx, `
|
|
INSERT INTO workspace_issue_topics (
|
|
topic_id, workspace_id, title, summary, is_default, pinned_at_unix_ms,
|
|
last_activity_at_unix_ms, created_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`, topic.TopicID, topic.WorkspaceID, topic.Title, topic.Summary, boolToInt(topic.IsDefault),
|
|
topic.PinnedAtUnixMS, topic.LastActivityAtUnixMS, topic.CreatedAtUnixMS, topic.UpdatedAtUnixMS)
|
|
if err != nil {
|
|
if isSQLiteUniqueConstraintError(err) {
|
|
return workspaceissues.Topic{}, workspaceissues.ErrTopicAlreadyExists
|
|
}
|
|
return workspaceissues.Topic{}, fmt.Errorf("create workspace issue topic: %w", err)
|
|
}
|
|
return s.GetTopic(ctx, topic.WorkspaceID, topic.TopicID)
|
|
}
|
|
|
|
func (s *SQLiteStore) GetTopic(ctx context.Context, workspaceID string, topicID string) (workspaceissues.Topic, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Topic{}, err
|
|
}
|
|
|
|
row := s.readDB.QueryRowContext(ctx, fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issue_topics
|
|
WHERE workspace_id = ? AND topic_id = ?
|
|
`, topicSelectColumns), workspaceID, topicID)
|
|
topic, err := scanWorkspaceIssueTopic(row)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return workspaceissues.Topic{}, workspaceissues.ErrTopicNotFound
|
|
}
|
|
if err != nil {
|
|
return workspaceissues.Topic{}, fmt.Errorf("get workspace issue topic: %w", err)
|
|
}
|
|
return topic, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) UpdateTopic(ctx context.Context, topic workspaceissues.Topic) (workspaceissues.Topic, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Topic{}, err
|
|
}
|
|
|
|
result, err := s.writeDB.ExecContext(ctx, `
|
|
UPDATE workspace_issue_topics
|
|
SET title = ?, summary = ?, pinned_at_unix_ms = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND topic_id = ?
|
|
`, topic.Title, topic.Summary, topic.PinnedAtUnixMS, topic.UpdatedAtUnixMS, topic.WorkspaceID, topic.TopicID)
|
|
if err != nil {
|
|
return workspaceissues.Topic{}, fmt.Errorf("update workspace issue topic: %w", err)
|
|
}
|
|
if err := requireRowsAffected(result, workspaceissues.ErrTopicNotFound, "update workspace issue topic"); err != nil {
|
|
return workspaceissues.Topic{}, err
|
|
}
|
|
return s.GetTopic(ctx, topic.WorkspaceID, topic.TopicID)
|
|
}
|
|
|
|
func (s *SQLiteStore) DeleteTopic(ctx context.Context, workspaceID string, topicID string) (bool, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return false, err
|
|
}
|
|
|
|
result, err := s.writeDB.ExecContext(ctx, `
|
|
DELETE FROM workspace_issue_topics
|
|
WHERE workspace_id = ? AND topic_id = ?
|
|
`, workspaceID, topicID)
|
|
if err != nil {
|
|
return false, fmt.Errorf("delete workspace issue topic: %w", err)
|
|
}
|
|
return rowsWereAffected(result, "delete workspace issue topic")
|
|
}
|
|
|
|
func (s *SQLiteStore) TouchTopicActivity(ctx context.Context, workspaceID string, topicID string, atUnixMS int64) error {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return err
|
|
}
|
|
result, err := s.writeDB.ExecContext(ctx, `
|
|
UPDATE workspace_issue_topics
|
|
SET last_activity_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND topic_id = ?
|
|
`, atUnixMS, workspaceID, topicID)
|
|
if err != nil {
|
|
return fmt.Errorf("touch workspace issue topic activity: %w", err)
|
|
}
|
|
return requireRowsAffected(result, workspaceissues.ErrTopicNotFound, "touch workspace issue topic activity")
|
|
}
|
|
|
|
func (s *SQLiteStore) ListIssues(ctx context.Context, filter workspaceissues.IssueListFilter) (workspaceissues.IssueList, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.IssueList{}, err
|
|
}
|
|
if err := s.ensureIssueWorkspace(ctx, filter.WorkspaceID); err != nil {
|
|
return workspaceissues.IssueList{}, err
|
|
}
|
|
|
|
where, args := issueListWhere(filter)
|
|
countWhere := strings.Join(where, " AND ")
|
|
countArgs := append([]any(nil), args...)
|
|
totalCount, err := s.countIssueRows(ctx, "workspace_issues", countWhere, countArgs)
|
|
if err != nil {
|
|
return workspaceissues.IssueList{}, err
|
|
}
|
|
statusCountWhere, statusCountArgs := issueListStatusCountWhere(filter)
|
|
counts, err := s.countIssueStatuses(
|
|
ctx,
|
|
"workspace_issues",
|
|
strings.Join(statusCountWhere, " AND "),
|
|
statusCountArgs,
|
|
)
|
|
if err != nil {
|
|
return workspaceissues.IssueList{}, err
|
|
}
|
|
|
|
if filter.Cursor != nil {
|
|
where = append(where, "(updated_at_unix_ms < ? OR (updated_at_unix_ms = ? AND id < ?))")
|
|
args = append(args, filter.Cursor.UpdatedAtUnixMS, filter.Cursor.UpdatedAtUnixMS, filter.Cursor.ID)
|
|
}
|
|
|
|
limit := issueListLimit(filter.PageSize, filter.ReturnAll)
|
|
query := fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issues
|
|
WHERE %s
|
|
ORDER BY updated_at_unix_ms DESC, id DESC%s
|
|
`, issueSelectColumns, strings.Join(where, " AND "), issueLimitClause(limit))
|
|
if limit > 0 {
|
|
args = append(args, limit)
|
|
}
|
|
|
|
rows, err := s.readDB.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return workspaceissues.IssueList{}, fmt.Errorf("list workspace issues: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
items, err := scanWorkspaceIssues(rows)
|
|
if err != nil {
|
|
return workspaceissues.IssueList{}, err
|
|
}
|
|
|
|
var nextCursor *workspaceissues.IssueListCursor
|
|
pageSize := normalizedIssuePageSize(filter.PageSize)
|
|
if !filter.ReturnAll && len(items) > pageSize {
|
|
last := items[pageSize-1]
|
|
nextCursor = &workspaceissues.IssueListCursor{
|
|
UpdatedAtUnixMS: last.UpdatedAtUnixMS,
|
|
ID: last.ID,
|
|
}
|
|
items = items[:pageSize]
|
|
}
|
|
|
|
return workspaceissues.IssueList{
|
|
Items: items,
|
|
NextCursor: nextCursor,
|
|
TotalCount: totalCount,
|
|
StatusCounts: counts,
|
|
}, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) CreateIssue(ctx context.Context, issue workspaceissues.Issue) (workspaceissues.Issue, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
if err := s.ensureIssueWorkspace(ctx, issue.WorkspaceID); err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
|
|
if _, err := insertWorkspaceIssue(ctx, s.writeDB, issue); err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
|
|
return s.GetIssue(ctx, issue.WorkspaceID, issue.IssueID)
|
|
}
|
|
|
|
func insertWorkspaceIssue(ctx context.Context, execer workspaceIssueExecer, issue workspaceissues.Issue) (workspaceissues.Issue, error) {
|
|
result, err := execer.ExecContext(ctx, `
|
|
INSERT INTO workspace_issues (
|
|
issue_id, topic_id, workspace_id, title, content, search_text, status,
|
|
planning_source, source_session_id, sequential_execution, parallel_execution, dispatch_paused, reasoning_intensity, orchestration_intensity,
|
|
budget_mode, budget_token_limit, budget_consumed_tokens,
|
|
budget_quota_waterline_percent, budget_remaining_quota_percent,
|
|
budget_has_remaining_quota, budget_status,
|
|
task_count, not_started_count, running_count, pending_acceptance_count,
|
|
completed_count, failed_count, canceled_count, creator_user_id,
|
|
creator_display_name, creator_avatar_url, created_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`, issue.IssueID, issue.TopicID, issue.WorkspaceID, issue.Title, issue.Content, issue.SearchText, string(issue.Status),
|
|
string(issue.PlanningSource), issue.SourceSessionID, boolToInt(issue.SequentialExecution), boolToInt(issue.ParallelExecution), boolToInt(issue.DispatchPaused), issue.ExecutionProfile.ReasoningIntensity,
|
|
issue.ExecutionProfile.OrchestrationIntensity, string(issue.Budget.Mode), issue.Budget.TokenLimit,
|
|
issue.Budget.ConsumedTokens, issue.Budget.QuotaWaterlinePercent, issue.Budget.RemainingQuotaPercent,
|
|
boolToInt(issue.Budget.HasRemainingQuota), string(issue.Budget.Status),
|
|
issue.TaskCount, issue.NotStartedCount, issue.RunningCount, issue.PendingAcceptanceCount,
|
|
issue.CompletedCount, issue.FailedCount, issue.CanceledCount, issue.CreatorUserID,
|
|
issue.CreatorDisplayName, issue.CreatorAvatarURL, issue.CreatedAtUnixMS, issue.UpdatedAtUnixMS)
|
|
if err != nil {
|
|
if isSQLiteUniqueConstraintError(err) {
|
|
return workspaceissues.Issue{}, workspaceissues.ErrIssueAlreadyExists
|
|
}
|
|
return workspaceissues.Issue{}, fmt.Errorf("create workspace issue: %w", err)
|
|
}
|
|
id, err := result.LastInsertId()
|
|
if err != nil {
|
|
return workspaceissues.Issue{}, fmt.Errorf("read created workspace issue id: %w", err)
|
|
}
|
|
issue.ID = uint64(id)
|
|
return issue, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) GetIssue(ctx context.Context, workspaceID string, issueID string) (workspaceissues.Issue, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
|
|
row := s.readDB.QueryRowContext(ctx, fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issues
|
|
WHERE workspace_id = ? AND issue_id = ?
|
|
`, issueSelectColumns), workspaceID, issueID)
|
|
issue, err := scanWorkspaceIssue(row)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return workspaceissues.Issue{}, workspaceissues.ErrIssueNotFound
|
|
}
|
|
if err != nil {
|
|
return workspaceissues.Issue{}, fmt.Errorf("get workspace issue: %w", err)
|
|
}
|
|
return issue, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) UpdateIssue(ctx context.Context, issue workspaceissues.Issue) (workspaceissues.Issue, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
|
|
result, err := s.writeDB.ExecContext(ctx, `
|
|
UPDATE workspace_issues
|
|
SET title = ?, content = ?, search_text = ?, status = ?,
|
|
dispatch_paused = ?, reasoning_intensity = ?, orchestration_intensity = ?, budget_mode = ?,
|
|
budget_token_limit = ?, budget_consumed_tokens = ?,
|
|
budget_quota_waterline_percent = ?, budget_remaining_quota_percent = ?,
|
|
budget_has_remaining_quota = ?, budget_status = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND issue_id = ?
|
|
`, issue.Title, issue.Content, issue.SearchText, string(issue.Status), boolToInt(issue.DispatchPaused), issue.ExecutionProfile.ReasoningIntensity,
|
|
issue.ExecutionProfile.OrchestrationIntensity, string(issue.Budget.Mode), issue.Budget.TokenLimit,
|
|
issue.Budget.ConsumedTokens, issue.Budget.QuotaWaterlinePercent, issue.Budget.RemainingQuotaPercent,
|
|
boolToInt(issue.Budget.HasRemainingQuota), string(issue.Budget.Status), issue.UpdatedAtUnixMS,
|
|
issue.WorkspaceID, issue.IssueID)
|
|
if err != nil {
|
|
return workspaceissues.Issue{}, fmt.Errorf("update workspace issue: %w", err)
|
|
}
|
|
if err := requireRowsAffected(result, workspaceissues.ErrIssueNotFound, "update workspace issue"); err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
return s.GetIssue(ctx, issue.WorkspaceID, issue.IssueID)
|
|
}
|
|
|
|
func (s *SQLiteStore) DeleteIssue(ctx context.Context, workspaceID string, issueID string, _ string) (bool, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return false, err
|
|
}
|
|
|
|
result, err := s.writeDB.ExecContext(ctx, `
|
|
DELETE FROM workspace_issues
|
|
WHERE workspace_id = ? AND issue_id = ?
|
|
`, workspaceID, issueID)
|
|
if err != nil {
|
|
return false, fmt.Errorf("delete workspace issue: %w", err)
|
|
}
|
|
return rowsWereAffected(result, "delete workspace issue")
|
|
}
|
|
|
|
func (s *SQLiteStore) RecalculateIssueProjection(ctx context.Context, workspaceID string, issueID string) (workspaceissues.Issue, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
|
|
counts, err := s.countIssueStatuses(
|
|
ctx,
|
|
"workspace_issue_tasks",
|
|
"workspace_id = ? AND issue_id = ? AND superseded_at_unix_ms = 0",
|
|
[]any{workspaceID, issueID},
|
|
)
|
|
if err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
status := workspaceissues.ProjectIssueStatus(counts)
|
|
now := unixMs(time.Now().UTC())
|
|
result, err := s.writeDB.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), counts.All, counts.NotStarted, counts.Running, counts.PendingAcceptance,
|
|
counts.Completed, counts.Failed, counts.Canceled, now, workspaceID, issueID)
|
|
if err != nil {
|
|
return workspaceissues.Issue{}, fmt.Errorf("recalculate workspace issue projection: %w", err)
|
|
}
|
|
if err := requireRowsAffected(result, workspaceissues.ErrIssueNotFound, "recalculate workspace issue projection"); err != nil {
|
|
return workspaceissues.Issue{}, err
|
|
}
|
|
return s.GetIssue(ctx, workspaceID, issueID)
|
|
}
|
|
|
|
func (s *SQLiteStore) AddContextRefs(ctx context.Context, refs []workspaceissues.ContextRef) ([]workspaceissues.ContextRef, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return nil, err
|
|
}
|
|
if len(refs) == 0 {
|
|
return nil, nil
|
|
}
|
|
if err := s.ensureContextRefParent(ctx, refs[0]); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
tx, err := s.writeDB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("begin workspace issue context refs transaction: %w", err)
|
|
}
|
|
defer func() {
|
|
_ = tx.Rollback()
|
|
}()
|
|
|
|
saved := make([]workspaceissues.ContextRef, 0, len(refs))
|
|
for _, ref := range refs {
|
|
created, err := insertWorkspaceIssueContextRef(ctx, tx, ref)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
saved = append(saved, created)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return nil, fmt.Errorf("commit workspace issue context refs: %w", err)
|
|
}
|
|
return saved, nil
|
|
}
|
|
|
|
func insertWorkspaceIssueContextRef(
|
|
ctx context.Context,
|
|
execer workspaceIssueExecer,
|
|
ref workspaceissues.ContextRef,
|
|
) (workspaceissues.ContextRef, error) {
|
|
result, err := execer.ExecContext(ctx, `
|
|
INSERT INTO workspace_issue_context_refs (
|
|
context_ref_id, workspace_id, issue_id, task_id, parent_kind,
|
|
ref_type, path, display_name, created_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`, ref.ContextRefID, ref.WorkspaceID, ref.IssueID, ref.TaskID, string(ref.ParentKind),
|
|
ref.RefType, ref.Path, ref.DisplayName, ref.CreatedAtUnixMS)
|
|
if err != nil {
|
|
if isSQLiteUniqueConstraintError(err) {
|
|
return workspaceissues.ContextRef{}, workspaceissues.ErrContextRefAlreadyExists
|
|
}
|
|
return workspaceissues.ContextRef{}, fmt.Errorf("add workspace issue context ref: %w", err)
|
|
}
|
|
id, err := result.LastInsertId()
|
|
if err != nil {
|
|
return workspaceissues.ContextRef{}, fmt.Errorf("read workspace issue context ref id: %w", err)
|
|
}
|
|
ref.ID = uint64(id)
|
|
return ref, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) ListContextRefs(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
taskID string,
|
|
parentKind workspaceissues.ContextRefParentKind,
|
|
) ([]workspaceissues.ContextRef, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rows, err := s.readDB.QueryContext(ctx, `
|
|
SELECT id, context_ref_id, workspace_id, issue_id, task_id, parent_kind,
|
|
ref_type, path, display_name, created_at_unix_ms
|
|
FROM workspace_issue_context_refs
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ? AND parent_kind = ?
|
|
ORDER BY created_at_unix_ms ASC, id ASC
|
|
`, workspaceID, issueID, taskID, string(parentKind))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list workspace issue context refs: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
items := make([]workspaceissues.ContextRef, 0)
|
|
for rows.Next() {
|
|
item, err := scanWorkspaceIssueContextRef(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate workspace issue context refs: %w", err)
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) GetIssueAttachmentContextRef(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
contextRefID string,
|
|
) (workspaceissues.ContextRef, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.ContextRef{}, err
|
|
}
|
|
item, err := scanWorkspaceIssueContextRef(s.readDB.QueryRowContext(ctx, `
|
|
SELECT id, context_ref_id, workspace_id, issue_id, task_id, parent_kind,
|
|
ref_type, path, display_name, created_at_unix_ms
|
|
FROM workspace_issue_context_refs
|
|
WHERE workspace_id = ? AND issue_id = ? AND context_ref_id = ?
|
|
`, strings.TrimSpace(workspaceID), strings.TrimSpace(issueID), strings.TrimSpace(contextRefID)))
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return workspaceissues.ContextRef{}, workspaceissues.ErrContextRefNotFound
|
|
}
|
|
if err != nil {
|
|
return workspaceissues.ContextRef{}, fmt.Errorf("get workspace issue attachment ContextRef: %w", err)
|
|
}
|
|
return item, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) HasIssueAttachmentReferencePath(ctx context.Context, path 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_context_refs
|
|
WHERE path = ?
|
|
LIMIT 1
|
|
`, path).Scan(&found)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
found = 0
|
|
} else if err != nil {
|
|
return false, fmt.Errorf("find workspace issue attachment ContextRef path: %w", err)
|
|
}
|
|
if found == 1 {
|
|
return true, nil
|
|
}
|
|
rows, err := s.readDB.QueryContext(ctx, `
|
|
SELECT payload_json
|
|
FROM workspace_issue_run_launch_intents
|
|
WHERE status IN ('prepared', 'leased')
|
|
`)
|
|
if err != nil {
|
|
return false, fmt.Errorf("list pending Issue Run attachment pins: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var payloadJSON string
|
|
if err := rows.Scan(&payloadJSON); err != nil {
|
|
return false, fmt.Errorf("scan pending Issue Run attachment pin: %w", err)
|
|
}
|
|
payload, err := workspacebiz.DecodeIssueRunLaunchPayload(payloadJSON)
|
|
if err != nil {
|
|
return false, fmt.Errorf("decode pending Issue Run attachment pins: %w", err)
|
|
}
|
|
for _, attachment := range payload.Attachments {
|
|
if attachment.Path == path {
|
|
return true, nil
|
|
}
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return false, fmt.Errorf("iterate pending Issue Run attachment pins: %w", err)
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) RemoveContextRef(
|
|
ctx context.Context,
|
|
workspaceID string,
|
|
issueID string,
|
|
taskID string,
|
|
parentKind workspaceissues.ContextRefParentKind,
|
|
contextRefID string,
|
|
) (bool, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return false, err
|
|
}
|
|
|
|
result, err := s.writeDB.ExecContext(ctx, `
|
|
DELETE FROM workspace_issue_context_refs
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ? AND parent_kind = ? AND context_ref_id = ?
|
|
`, workspaceID, issueID, taskID, string(parentKind), contextRefID)
|
|
if err != nil {
|
|
return false, fmt.Errorf("remove workspace issue context ref: %w", err)
|
|
}
|
|
return rowsWereAffected(result, "remove workspace issue context ref")
|
|
}
|
|
|
|
func (s *SQLiteStore) CreateRun(ctx context.Context, run workspaceissues.Run) (workspaceissues.Run, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Run{}, err
|
|
}
|
|
if run.TaskID != "" {
|
|
if _, err := s.GetTask(ctx, run.WorkspaceID, run.IssueID, run.TaskID); err != nil {
|
|
return workspaceissues.Run{}, err
|
|
}
|
|
} else if _, err := s.GetIssue(ctx, run.WorkspaceID, run.IssueID); err != nil {
|
|
return workspaceissues.Run{}, err
|
|
}
|
|
|
|
if err := insertWorkspaceIssueRun(ctx, s.writeDB, run); err != nil {
|
|
return workspaceissues.Run{}, err
|
|
}
|
|
return s.GetRun(ctx, run.WorkspaceID, run.IssueID, run.TaskID, run.RunID)
|
|
}
|
|
|
|
func insertWorkspaceIssueRun(ctx context.Context, execer workspaceIssueExecer, run workspaceissues.Run) error {
|
|
_, err := execer.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 workspaceissues.ErrRunAlreadyExists
|
|
}
|
|
return fmt.Errorf("create workspace issue run: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *SQLiteStore) CompleteRun(ctx context.Context, run workspaceissues.Run, outputs []workspaceissues.RunOutput) (workspaceissues.Run, []workspaceissues.RunOutput, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Run{}, nil, err
|
|
}
|
|
|
|
tx, err := s.writeDB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return workspaceissues.Run{}, nil, fmt.Errorf("begin workspace issue run transaction: %w", err)
|
|
}
|
|
defer func() {
|
|
_ = tx.Rollback()
|
|
}()
|
|
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_issue_runs
|
|
SET status = ?, input_tokens = ?, output_tokens = ?, cache_read_tokens = ?,
|
|
cache_write_tokens = ?, summary = ?, error_message = ?,
|
|
completed_at_unix_ms = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ? AND run_id = ?
|
|
`, string(run.Status), run.Usage.InputTokens, run.Usage.OutputTokens, run.Usage.CacheReadTokens,
|
|
run.Usage.CacheWriteTokens, run.Summary, run.ErrorMessage, run.CompletedAtUnixMS, run.UpdatedAtUnixMS,
|
|
run.WorkspaceID, run.IssueID, run.TaskID, run.RunID)
|
|
if err != nil {
|
|
return workspaceissues.Run{}, nil, fmt.Errorf("complete workspace issue run: %w", err)
|
|
}
|
|
if err := requireRowsAffected(result, workspaceissues.ErrRunNotFound, "complete workspace issue run"); err != nil {
|
|
return workspaceissues.Run{}, nil, err
|
|
}
|
|
|
|
if _, err := tx.ExecContext(ctx, `
|
|
DELETE FROM workspace_issue_run_outputs
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ? AND run_id = ?
|
|
`, run.WorkspaceID, run.IssueID, run.TaskID, run.RunID); err != nil {
|
|
return workspaceissues.Run{}, nil, fmt.Errorf("replace workspace issue run outputs: %w", err)
|
|
}
|
|
|
|
saved := make([]workspaceissues.RunOutput, 0, len(outputs))
|
|
for _, output := range outputs {
|
|
result, err := tx.ExecContext(ctx, `
|
|
INSERT INTO workspace_issue_run_outputs (
|
|
output_id, run_id, task_id, issue_id, workspace_id, path, display_name,
|
|
media_type, size_bytes, created_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`, output.OutputID, output.RunID, output.TaskID, output.IssueID, output.WorkspaceID,
|
|
output.Path, output.DisplayName, output.MediaType, output.SizeBytes, output.CreatedAtUnixMS)
|
|
if err != nil {
|
|
return workspaceissues.Run{}, nil, fmt.Errorf("add workspace issue run output: %w", err)
|
|
}
|
|
id, err := result.LastInsertId()
|
|
if err != nil {
|
|
return workspaceissues.Run{}, nil, fmt.Errorf("read workspace issue run output id: %w", err)
|
|
}
|
|
output.ID = uint64(id)
|
|
saved = append(saved, output)
|
|
}
|
|
|
|
if err := tx.Commit(); err != nil {
|
|
return workspaceissues.Run{}, nil, fmt.Errorf("commit workspace issue run: %w", err)
|
|
}
|
|
|
|
completed, err := s.GetRun(ctx, run.WorkspaceID, run.IssueID, run.TaskID, run.RunID)
|
|
if err != nil {
|
|
return workspaceissues.Run{}, nil, err
|
|
}
|
|
return completed, saved, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) ListRuns(ctx context.Context, workspaceID string, issueID string, taskID string) ([]workspaceissues.Run, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
query := fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issue_runs
|
|
ORDER BY created_at_unix_ms DESC, id DESC
|
|
`, runSelectColumns)
|
|
args := []any{workspaceID}
|
|
where := "WHERE workspace_id = ?"
|
|
if issueID != "" {
|
|
where += " AND issue_id = ?"
|
|
args = append(args, issueID)
|
|
}
|
|
if taskID != "" {
|
|
where += " AND task_id = ?"
|
|
args = append(args, taskID)
|
|
}
|
|
query = strings.Replace(query, "FROM workspace_issue_runs\n", "FROM workspace_issue_runs\n"+where+"\n", 1)
|
|
rows, err := s.readDB.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list workspace issue runs: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
items := make([]workspaceissues.Run, 0)
|
|
for rows.Next() {
|
|
item, err := scanWorkspaceIssueRun(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate workspace issue runs: %w", err)
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) ListRunningRuns(ctx context.Context, workspaceID string, limit int) ([]workspaceissues.Run, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return nil, err
|
|
}
|
|
if limit <= 0 {
|
|
limit = 100
|
|
}
|
|
|
|
rows, err := s.readDB.QueryContext(ctx, fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issue_runs
|
|
WHERE workspace_id = ? AND status = ? AND TRIM(agent_session_id) <> ''
|
|
ORDER BY updated_at_unix_ms ASC, id ASC
|
|
LIMIT ?
|
|
`, runSelectColumns), workspaceID, string(workspaceissues.StatusRunning), limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list running workspace issue runs: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
items := make([]workspaceissues.Run, 0)
|
|
for rows.Next() {
|
|
item, err := scanWorkspaceIssueRun(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate running workspace issue runs: %w", err)
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) GetRun(ctx context.Context, workspaceID string, issueID string, taskID string, runID string) (workspaceissues.Run, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return workspaceissues.Run{}, err
|
|
}
|
|
|
|
row := s.readDB.QueryRowContext(ctx, fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issue_runs
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ? AND run_id = ?
|
|
`, runSelectColumns), workspaceID, issueID, taskID, runID)
|
|
run, err := scanWorkspaceIssueRun(row)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return workspaceissues.Run{}, workspaceissues.ErrRunNotFound
|
|
}
|
|
if err != nil {
|
|
return workspaceissues.Run{}, fmt.Errorf("get workspace issue run: %w", err)
|
|
}
|
|
return run, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) ListRunOutputs(ctx context.Context, workspaceID string, issueID string, taskID string, runID string) ([]workspaceissues.RunOutput, error) {
|
|
if err := s.ensureIssueDatabase(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rows, err := s.readDB.QueryContext(ctx, fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM workspace_issue_run_outputs
|
|
WHERE workspace_id = ? AND issue_id = ? AND task_id = ? AND run_id = ?
|
|
ORDER BY created_at_unix_ms ASC, id ASC
|
|
`, runOutputSelectColumns), workspaceID, issueID, taskID, runID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list workspace issue run outputs: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
outputs := make([]workspaceissues.RunOutput, 0)
|
|
for rows.Next() {
|
|
item, err := scanWorkspaceIssueRunOutput(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
outputs = append(outputs, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate workspace issue run outputs: %w", err)
|
|
}
|
|
return outputs, nil
|
|
}
|
|
|
|
func (s *SQLiteStore) ListLatestRunOutputs(ctx context.Context, workspaceID string, issueID string, taskID string) ([]workspaceissues.RunOutput, error) {
|
|
runs, err := s.ListRuns(ctx, workspaceID, issueID, taskID)
|
|
if err != nil || len(runs) == 0 {
|
|
return nil, err
|
|
}
|
|
return s.ListRunOutputs(ctx, workspaceID, issueID, runs[0].TaskID, runs[0].RunID)
|
|
}
|