Add review workflows, tree-shaped workflows and a workflow editor
This commit is contained in:
@@ -10,18 +10,26 @@ import (
|
||||
"g.fc.run/ai/byre/internal/api"
|
||||
)
|
||||
|
||||
const reviewColumns = `id, task_id, snapshot_n, coalesce(agent_session_id, ''), vendor, model, state, result, error, created_at, finished_at, finding_states, finding_responses`
|
||||
const reviewColumns = `id, task_id, snapshot_n, stage, coalesce(agent_session_id, ''), vendor, model, state, result, error, created_at, finished_at, finding_states, finding_responses`
|
||||
|
||||
func (s *Store) CreateReview(ctx context.Context, r api.Review) error {
|
||||
func (s *Store) CreateReview(ctx context.Context, r api.Review, def *api.WorkflowDefinition) error {
|
||||
finishedAt := ""
|
||||
if r.FinishedAt != nil {
|
||||
finishedAt = formatTime(*r.FinishedAt)
|
||||
}
|
||||
defText := ""
|
||||
if def != nil {
|
||||
b, err := json.Marshal(def)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defText = string(b)
|
||||
}
|
||||
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
_, err := tx.ExecContext(ctx, `
|
||||
INSERT INTO reviews (id, task_id, snapshot_n, agent_session_id, vendor, model, state, created_at, finished_at)
|
||||
VALUES (?, ?, ?, nullif(?, ''), ?, ?, ?, ?, ?)
|
||||
`, r.Id, r.TaskId, r.SnapshotN, r.AgentSessionId, r.Vendor, r.Model, string(r.State), formatTime(r.CreatedAt), finishedAt)
|
||||
INSERT INTO reviews (id, task_id, snapshot_n, stage, definition, agent_session_id, vendor, model, state, error, created_at, finished_at)
|
||||
VALUES (?, ?, ?, ?, ?, nullif(?, ''), ?, ?, ?, ?, ?, ?)
|
||||
`, r.Id, r.TaskId, r.SnapshotN, r.Stage, defText, r.AgentSessionId, r.Vendor, r.Model, string(r.State), r.Error, formatTime(r.CreatedAt), finishedAt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -32,6 +40,23 @@ func (s *Store) CreateReview(ctx context.Context, r api.Review) error {
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Store) ReviewDefinition(ctx context.Context, id string) (api.WorkflowDefinition, error) {
|
||||
text := ""
|
||||
err := s.db.QueryRowContext(ctx, `SELECT definition FROM reviews WHERE id = ?`, id).Scan(&text)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return api.WorkflowDefinition{}, ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return api.WorkflowDefinition{}, err
|
||||
}
|
||||
def := api.WorkflowDefinition{}
|
||||
if text == "" {
|
||||
return def, nil
|
||||
}
|
||||
err = json.Unmarshal([]byte(text), &def)
|
||||
return def, err
|
||||
}
|
||||
|
||||
func (s *Store) GetReview(ctx context.Context, id string) (api.Review, error) {
|
||||
r, err := scanReview(s.db.QueryRowContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE id = ?`, id))
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
@@ -49,7 +74,7 @@ func (s *Store) ReviewBySession(ctx context.Context, sessionID string) (api.Revi
|
||||
}
|
||||
|
||||
func (s *Store) ListReviews(ctx context.Context, taskID string) ([]api.Review, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE task_id = ? ORDER BY snapshot_n`, taskID)
|
||||
rows, err := s.db.QueryContext(ctx, `SELECT `+reviewColumns+` FROM reviews WHERE task_id = ? ORDER BY snapshot_n, created_at`, taskID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -64,6 +89,29 @@ func (s *Store) RunningReviews(ctx context.Context) ([]api.Review, error) {
|
||||
return scanReviews(rows)
|
||||
}
|
||||
|
||||
func (s *Store) LatestDoneStages(ctx context.Context) ([]string, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT r.id FROM reviews r JOIN tasks t ON t.id = r.task_id
|
||||
WHERE t.status = ? AND r.state = ? AND r.definition != ''
|
||||
AND r.snapshot_n = (SELECT max(snapshot_n) FROM reviews WHERE task_id = r.task_id)
|
||||
ORDER BY r.created_at`, string(api.TaskStatusActive), string(api.ReviewStateDone))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
ids := []string{}
|
||||
for rows.Next() {
|
||||
id := ""
|
||||
err = rows.Scan(&id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Store) FinishReview(ctx context.Context, id string, state api.ReviewState, result map[string]any, errText string) error {
|
||||
text := ""
|
||||
if result != nil {
|
||||
@@ -205,7 +253,7 @@ func scanReview(row scanner) (api.Review, error) {
|
||||
finishedAt := ""
|
||||
findingStates := ""
|
||||
findingResponses := ""
|
||||
err := row.Scan(&r.Id, &r.TaskId, &r.SnapshotN, &r.AgentSessionId, &r.Vendor, &r.Model, &state, &result, &r.Error, &createdAt, &finishedAt, &findingStates, &findingResponses)
|
||||
err := row.Scan(&r.Id, &r.TaskId, &r.SnapshotN, &r.Stage, &r.AgentSessionId, &r.Vendor, &r.Model, &state, &result, &r.Error, &createdAt, &finishedAt, &findingStates, &findingResponses)
|
||||
if err != nil {
|
||||
return r, err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user