internal/storage: SQLite-БД для задач и трассировок (таски + статус-машина, трассы с группировкой по task_id)
All checks were successful
build-test / build (push) Successful in 30s

This commit is contained in:
Hermes
2026-08-14 23:38:57 +05:00
parent cc180d7517
commit 089e196dbc
7 changed files with 821 additions and 0 deletions

View File

@@ -0,0 +1,21 @@
package storage
import "errors"
// Классы ошибок Ratatoskr-стиля: S1S5
var (
// S1 — сущность не найдена
ErrNotFound = errors.New("S1: task not found")
// S2 — дубликат (task_tag, title+chat_id)
ErrDuplicate = errors.New("S2: duplicate entity")
// S3 — невалидный переход статуса
ErrInvalidStatus = errors.New("S3: invalid status transition")
// S4 — внутренняя ошибка БД
ErrDB = errors.New("S4: database error")
// S5 — трасса не найдена
ErrTraceNotFound = errors.New("S5: trace not found")
)

104
internal/storage/models.go Normal file
View File

@@ -0,0 +1,104 @@
package storage
// Status — статус задачи (state machine).
type Status string
const (
StatusDraft Status = "draft" // только что создана
StatusCollecting Status = "collecting" // аналитик собирает детали
StatusReady Status = "ready" // черновик готов, ждёт запуска
StatusRunning Status = "running" // opencode работает
StatusSuccess Status = "success" // задача выполнена
StatusFailed Status = "failed" // ошибка выполнения
StatusTimeout Status = "timeout" // таймаут opencode
StatusCancelled Status = "cancelled" // отменена пользователем
StatusAborted Status = "aborted" // сбой сбора, черновик выброшен
StatusClosed Status = "closed" // закрыта вручную
)
// AllStatuses — все возможные статусы для валидации.
var AllStatuses = []Status{
StatusDraft, StatusCollecting, StatusReady,
StatusRunning, StatusSuccess, StatusFailed, StatusTimeout,
StatusCancelled, StatusAborted, StatusClosed,
}
// validTransitions задаёт разрешённые переходы статусов.
var validTransitions = map[Status][]Status{
StatusDraft: {StatusCollecting, StatusCancelled, StatusAborted},
StatusCollecting: {StatusReady, StatusDraft, StatusCancelled, StatusAborted},
StatusReady: {StatusRunning, StatusCancelled, StatusAborted, StatusClosed},
StatusRunning: {StatusSuccess, StatusFailed, StatusTimeout, StatusCancelled},
StatusSuccess: {StatusClosed},
StatusFailed: {StatusReady, StatusClosed, StatusCancelled}, // retry
StatusTimeout: {StatusReady, StatusClosed, StatusCancelled}, // retry
StatusCancelled: {StatusClosed},
StatusAborted: {StatusClosed},
StatusClosed: {}, // терминальный
}
// IsValidTransition проверяет, допустим ли переход from → to.
func IsValidTransition(from, to Status) bool {
allowed, ok := validTransitions[from]
if !ok {
return false
}
for _, s := range allowed {
if s == to {
return true
}
}
return false
}
// IsTerminal возвращает true, если статус терминальный.
func IsTerminal(s Status) bool {
return s == StatusSuccess || s == StatusCancelled ||
s == StatusAborted || s == StatusClosed
}
// Task — запись задачи в БД.
type Task struct {
ID int64 `json:"id"`
ChatID string `json:"chat_id"` // tg://<id>
Title string `json:"title"`
Goal string `json:"goal"`
Repo string `json:"repo"`
Why string `json:"why"`
AC string `json:"ac"` // acceptance criteria
TaskTag string `json:"task_tag"` // UUID, стабильный на всю жизнь
Status Status `json:"status"`
CreatedAt SQLiteTime `json:"created_at"`
UpdatedAt SQLiteTime `json:"updated_at"`
}
// Trace — запись трассировки выполнения.
type TraceStatus string
const (
TraceRunning TraceStatus = "running"
TraceSuccess TraceStatus = "success"
TraceFailed TraceStatus = "failed"
TraceTimeout TraceStatus = "timeout"
)
// Trace — лог одного субагента.
type Trace struct {
ID int64 `json:"id"`
TaskID int64 `json:"task_id"`
Agent string `json:"agent"` // analyst | researcher | dev | reviewer
SessionID string `json:"session_id"` // opencode session_id
Prompt string `json:"prompt"`
Output string `json:"output"` // полный NDJSON или summary
Status TraceStatus `json:"status"`
StartedAt SQLiteTime `json:"started_at"`
FinishedAt NullSQLiteTime `json:"finished_at,omitempty"`
}
// TaskFilter — параметры фильтрации списка задач.
type TaskFilter struct {
ChatID string
Status Status
Limit int
Offset int
}

View File

@@ -0,0 +1,49 @@
package storage
import (
"testing"
)
func TestIsValidTransition(t *testing.T) {
tests := []struct {
from, to Status
want bool
}{
{StatusDraft, StatusCollecting, true},
{StatusDraft, StatusCancelled, true},
{StatusDraft, StatusRunning, false},
{StatusDraft, StatusClosed, false},
{StatusRunning, StatusSuccess, true},
{StatusRunning, StatusFailed, true},
{StatusRunning, StatusTimeout, true},
{StatusSuccess, StatusClosed, true},
{StatusSuccess, StatusDraft, false},
{StatusFailed, StatusReady, true}, // retry
{StatusFailed, StatusClosed, true},
{StatusFailed, StatusRunning, false},
{StatusTimeout, StatusReady, true}, // retry
{StatusClosed, StatusDraft, false},
{StatusClosed, StatusRunning, false},
}
for _, tc := range tests {
got := IsValidTransition(tc.from, tc.to)
if got != tc.want {
t.Errorf("IsValidTransition(%q → %q) = %v, want %v", tc.from, tc.to, got, tc.want)
}
}
}
func TestIsTerminal(t *testing.T) {
if !IsTerminal(StatusSuccess) {
t.Error("StatusSuccess should be terminal")
}
if !IsTerminal(StatusCancelled) {
t.Error("StatusCancelled should be terminal")
}
if IsTerminal(StatusDraft) {
t.Error("StatusDraft should NOT be terminal")
}
if IsTerminal(StatusRunning) {
t.Error("StatusRunning should NOT be terminal")
}
}

158
internal/storage/storage.go Normal file
View File

@@ -0,0 +1,158 @@
package storage
import (
"context"
"database/sql"
"database/sql/driver"
"fmt"
"time"
_ "modernc.org/sqlite"
)
// SQLiteTime — time.Time с sql.Scanner и driver.Valuer для modernc.org/sqlite.
// Хранит как INTEGER (Unix-секунды).
type SQLiteTime time.Time
func (st *SQLiteTime) Scan(src any) error {
if src == nil {
*st = SQLiteTime(time.Time{})
return nil
}
switch v := src.(type) {
case int64:
*st = SQLiteTime(time.Unix(v, 0).UTC())
return nil
case float64:
*st = SQLiteTime(time.Unix(int64(v), 0).UTC())
return nil
case string:
// fallback: RFC3339
t, err := time.Parse(time.RFC3339, v)
if err != nil {
return fmt.Errorf("parse time %q: %w", v, err)
}
*st = SQLiteTime(t)
return nil
default:
return fmt.Errorf("cannot scan %T as SQLiteTime", src)
}
}
func (st SQLiteTime) Value() (driver.Value, error) {
return time.Time(st).Unix(), nil
}
func (st SQLiteTime) Time() time.Time {
return time.Time(st)
}
// NullSQLiteTime — nullable версия SQLiteTime.
type NullSQLiteTime struct {
Time SQLiteTime
Valid bool
}
func (n *NullSQLiteTime) Scan(src any) error {
if src == nil {
n.Valid = false
return nil
}
n.Valid = true
return n.Time.Scan(src)
}
func (n NullSQLiteTime) Value() (driver.Value, error) {
if !n.Valid {
return nil, nil
}
return n.Time.Value()
}
// Storage — компонент БД для задач и трассировок.
type Storage struct {
db *sql.DB
}
// Open открывает SQLite-БД и выполняет миграции.
// path — путь к файлу БД; ":memory:" для тестов.
func Open(ctx context.Context, path string) (*Storage, error) {
db, err := sql.Open("sqlite", path)
if err != nil {
return nil, fmt.Errorf("%w: open: %w", ErrDB, err)
}
// Прагмы: WAL + синхронность
pragmas := []string{
"PRAGMA journal_mode=WAL",
"PRAGMA synchronous=NORMAL",
"PRAGMA busy_timeout=5000",
"PRAGMA foreign_keys=ON",
}
for _, p := range pragmas {
if _, err := db.ExecContext(ctx, p); err != nil {
db.Close()
return nil, fmt.Errorf("%w: pragma %q: %w", ErrDB, p, err)
}
}
s := &Storage{db: db}
if err := s.migrate(ctx); err != nil {
db.Close()
return nil, err
}
return s, nil
}
// Close закрывает БД.
func (s *Storage) Close() error {
return s.db.Close()
}
// DB возвращает сырой *sql.DB для использования в транзакциях.
func (s *Storage) DB() *sql.DB {
return s.db
}
// migrate создаёт таблицы при первом запуске.
func (s *Storage) migrate(ctx context.Context) error {
schema := `
CREATE TABLE IF NOT EXISTS tasks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
chat_id TEXT NOT NULL DEFAULT '',
title TEXT NOT NULL DEFAULT '',
goal TEXT NOT NULL DEFAULT '',
repo TEXT NOT NULL DEFAULT '',
why TEXT NOT NULL DEFAULT '',
ac TEXT NOT NULL DEFAULT '',
task_tag TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'draft',
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
updated_at INTEGER NOT NULL DEFAULT (unixepoch())
);
CREATE INDEX IF NOT EXISTS idx_tasks_chat ON tasks(chat_id);
CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
CREATE TABLE IF NOT EXISTS traces (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
agent TEXT NOT NULL DEFAULT '',
session_id TEXT NOT NULL DEFAULT '',
prompt TEXT NOT NULL DEFAULT '',
output TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'running',
started_at INTEGER NOT NULL DEFAULT (unixepoch()),
finished_at INTEGER
);
CREATE INDEX IF NOT EXISTS idx_traces_task ON traces(task_id);
`
if _, err := s.db.ExecContext(ctx, schema); err != nil {
return fmt.Errorf("%w: migrate: %w", ErrDB, err)
}
return nil
}
// Now возвращает текущее время UTC как SQLiteTime.
func Now() SQLiteTime {
return SQLiteTime(time.Now().UTC())
}

View File

@@ -0,0 +1,239 @@
package storage
import (
"context"
"errors"
"testing"
)
func setupTestDB(t *testing.T) (*Storage, context.Context) {
t.Helper()
ctx := context.Background()
s, err := Open(ctx, ":memory:")
if err != nil {
t.Fatalf("Open(:memory:): %v", err)
}
t.Cleanup(func() { s.Close() })
return s, ctx
}
func TestCreateAndGetTask(t *testing.T) {
s, ctx := setupTestDB(t)
task := &Task{
ChatID: "tg://123",
Title: "Test task",
Goal: "Test goal",
Repo: "test/repo",
TaskTag: "abc-123",
}
id, err := s.CreateTask(ctx, task)
if err != nil {
t.Fatalf("CreateTask: %v", err)
}
if id == 0 {
t.Fatal("expected non-zero ID")
}
if task.Status != StatusDraft {
t.Fatalf("expected draft status, got %s", task.Status)
}
got, err := s.GetTask(ctx, id)
if err != nil {
t.Fatalf("GetTask: %v", err)
}
if got.Title != "Test task" {
t.Fatalf("title = %q, want %q", got.Title, "Test task")
}
if got.Status != StatusDraft {
t.Fatalf("status = %s, want draft", got.Status)
}
}
func TestGetTaskNotFound(t *testing.T) {
s, ctx := setupTestDB(t)
_, err := s.GetTask(ctx, 999)
if !IsNotFound(err) {
t.Fatalf("expected ErrNotFound, got %v", err)
}
}
func TestUpdateTaskStatus(t *testing.T) {
s, ctx := setupTestDB(t)
task := &Task{
ChatID: "tg://123",
Title: "Update test",
TaskTag: "def-456",
}
_, _ = s.CreateTask(ctx, task)
// draft → collecting — валидно
task.Status = StatusCollecting
if err := s.UpdateTask(ctx, task); err != nil {
t.Fatalf("UpdateTask draft→collecting: %v", err)
}
got, _ := s.GetTask(ctx, task.ID)
if got.Status != StatusCollecting {
t.Fatalf("status = %s, want collecting", got.Status)
}
// collecting → ready
task.Status = StatusReady
if err := s.UpdateTask(ctx, task); err != nil {
t.Fatalf("UpdateTask collecting→ready: %v", err)
}
// ready → running
task.Status = StatusRunning
if err := s.UpdateTask(ctx, task); err != nil {
t.Fatalf("UpdateTask ready→running: %v", err)
}
// running → success
task.Status = StatusSuccess
if err := s.UpdateTask(ctx, task); err != nil {
t.Fatalf("UpdateTask running→success: %v", err)
}
}
func TestUpdateTaskInvalidTransition(t *testing.T) {
s, ctx := setupTestDB(t)
task := &Task{ChatID: "tg://1", TaskTag: "invalid"}
_, _ = s.CreateTask(ctx, task)
task.Status = StatusRunning // draft → running невалидно
err := s.UpdateTask(ctx, task)
if !IsInvalidStatus(err) {
t.Fatalf("expected ErrInvalidStatus, got %v", err)
}
}
func TestUpdateTaskNotFound(t *testing.T) {
s, ctx := setupTestDB(t)
err := s.UpdateTask(ctx, &Task{ID: 999, Status: StatusReady})
if !IsNotFound(err) {
t.Fatalf("expected ErrNotFound, got %v", err)
}
}
func TestListTasks(t *testing.T) {
s, ctx := setupTestDB(t)
for i := 0; i < 5; i++ {
_, _ = s.CreateTask(ctx, &Task{
ChatID: "tg://user",
Title: "Task",
TaskTag: "tag",
})
}
tasks, err := s.ListTasks(ctx, TaskFilter{ChatID: "tg://user", Limit: 10})
if err != nil {
t.Fatalf("ListTasks: %v", err)
}
if len(tasks) != 5 {
t.Fatalf("expected 5 tasks, got %d", len(tasks))
}
}
func TestDeleteTask(t *testing.T) {
s, ctx := setupTestDB(t)
task := &Task{ChatID: "tg://del", TaskTag: "delete-me"}
id, _ := s.CreateTask(ctx, task)
if err := s.DeleteTask(ctx, id); err != nil {
t.Fatalf("DeleteTask: %v", err)
}
_, err := s.GetTask(ctx, id)
if !IsNotFound(err) {
t.Fatal("expected ErrNotFound after delete")
}
}
func TestAppendAndGetTraces(t *testing.T) {
s, ctx := setupTestDB(t)
task := &Task{ChatID: "tg://trace", TaskTag: "trace-test"}
id, _ := s.CreateTask(ctx, task)
tr := &Trace{
TaskID: id,
Agent: "analyst",
SessionID: "sess-001",
Prompt: "analyze this",
Output: "",
}
traceID, err := s.AppendTrace(ctx, tr)
if err != nil {
t.Fatalf("AppendTrace: %v", err)
}
if traceID == 0 {
t.Fatal("expected non-zero trace ID")
}
if tr.Status != TraceRunning {
t.Fatalf("trace status = %s, want running", tr.Status)
}
// апдейт статуса
if err := s.UpdateTraceStatus(ctx, traceID, TraceSuccess); err != nil {
t.Fatalf("UpdateTraceStatus: %v", err)
}
// добавим вторую трассу
_, _ = s.AppendTrace(ctx, &Trace{
TaskID: id,
Agent: "dev",
SessionID: "sess-002",
})
traces, err := s.GetTraces(ctx, id)
if err != nil {
t.Fatalf("GetTraces: %v", err)
}
if len(traces) != 2 {
t.Fatalf("expected 2 traces, got %d", len(traces))
}
if traces[0].Status != TraceSuccess {
t.Fatalf("first trace status = %s, want success", traces[0].Status)
}
if traces[1].Agent != "dev" {
t.Fatalf("second trace agent = %s, want dev", traces[1].Agent)
}
}
func TestGetLatestTrace(t *testing.T) {
s, ctx := setupTestDB(t)
task := &Task{ChatID: "tg://latest", TaskTag: "latest-trace"}
id, _ := s.CreateTask(ctx, task)
_, _ = s.AppendTrace(ctx, &Trace{TaskID: id, Agent: "analyst", SessionID: "s1"})
_, _ = s.AppendTrace(ctx, &Trace{TaskID: id, Agent: "analyst", SessionID: "s2"})
latest, err := s.GetLatestTrace(ctx, id, "analyst")
if err != nil {
t.Fatalf("GetLatestTrace: %v", err)
}
if latest.SessionID != "s2" {
t.Fatalf("latest session = %s, want s2", latest.SessionID)
}
}
func TestUpdateTraceOutput(t *testing.T) {
s, ctx := setupTestDB(t)
task := &Task{ChatID: "tg://out", TaskTag: "output"}
id, _ := s.CreateTask(ctx, task)
traceID, _ := s.AppendTrace(ctx, &Trace{TaskID: id, Agent: "dev"})
if err := s.UpdateTraceOutput(ctx, traceID, "hello world"); err != nil {
t.Fatalf("UpdateTraceOutput: %v", err)
}
traces, _ := s.GetTraces(ctx, id)
if traces[0].Output != "hello world" {
t.Fatalf("output = %q, want %q", traces[0].Output, "hello world")
}
}
// хелперы для проверки классов ошибок
func IsNotFound(err error) bool {
return errors.Is(err, ErrNotFound)
}
func IsInvalidStatus(err error) bool {
return errors.Is(err, ErrInvalidStatus)
}

138
internal/storage/tasks.go Normal file
View File

@@ -0,0 +1,138 @@
package storage
import (
"context"
"database/sql"
"fmt"
)
// CreateTask создаёт задачу со статусом draft. Возвращает её ID.
// chat_id и task_tag передаются извне (в диалоге — chat.Address + uuid).
func (s *Storage) CreateTask(ctx context.Context, t *Task) (int64, error) {
now := Now()
res, err := s.db.ExecContext(ctx, `
INSERT INTO tasks (chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
t.ChatID, t.Title, t.Goal, t.Repo, t.Why, t.AC,
t.TaskTag, StatusDraft, now, now,
)
if err != nil {
return 0, fmt.Errorf("%w: create task: %w", ErrDB, err)
}
id, err := res.LastInsertId()
if err != nil {
return 0, fmt.Errorf("%w: last insert id: %w", ErrDB, err)
}
t.ID = id
t.Status = StatusDraft
t.CreatedAt = now
t.UpdatedAt = now
return id, nil
}
// GetTask возвращает задачу по ID.
func (s *Storage) GetTask(ctx context.Context, id int64) (*Task, error) {
t := &Task{}
err := s.db.QueryRowContext(ctx, `
SELECT id, chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at
FROM tasks WHERE id = ?`, id).Scan(
&t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo,
&t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt,
)
if err == sql.ErrNoRows {
return nil, fmt.Errorf("%w: task %d", ErrNotFound, id)
}
if err != nil {
return nil, fmt.Errorf("%w: get task %d: %w", ErrDB, id, err)
}
return t, nil
}
// UpdateTask обновляет поля задачи. Если newStatus указан и отличен от текущего —
// проверяет валидность перехода. Возвращает ошибку ErrInvalidStatus при недопустимом переходе.
func (s *Storage) UpdateTask(ctx context.Context, t *Task) error {
// получаем текущий статус для валидации перехода
var oldStatus Status
err := s.db.QueryRowContext(ctx, `SELECT status FROM tasks WHERE id = ?`, t.ID).Scan(&oldStatus)
if err == sql.ErrNoRows {
return fmt.Errorf("%w: task %d", ErrNotFound, t.ID)
}
if err != nil {
return fmt.Errorf("%w: get status %d: %w", ErrDB, t.ID, err)
}
if t.Status != "" && t.Status != oldStatus {
if !IsValidTransition(oldStatus, t.Status) {
return fmt.Errorf("%w: %s → %s for task %d", ErrInvalidStatus, oldStatus, t.Status, t.ID)
}
} else if t.Status == "" {
t.Status = oldStatus
}
now := Now()
res, err := s.db.ExecContext(ctx, `
UPDATE tasks
SET title=?, goal=?, repo=?, why=?, ac=?, status=?, updated_at=?
WHERE id=?`,
t.Title, t.Goal, t.Repo, t.Why, t.AC, t.Status, now, t.ID,
)
if err != nil {
return fmt.Errorf("%w: update task %d: %w", ErrDB, t.ID, err)
}
n, _ := res.RowsAffected()
if n == 0 {
return fmt.Errorf("%w: task %d", ErrNotFound, t.ID)
}
t.UpdatedAt = now
return nil
}
// ListTasks возвращает список задач с фильтрацией.
func (s *Storage) ListTasks(ctx context.Context, filter TaskFilter) ([]*Task, error) {
if filter.Limit <= 0 {
filter.Limit = 50
}
where := "1=1"
args := []any{}
if filter.ChatID != "" {
where += " AND chat_id = ?"
args = append(args, filter.ChatID)
}
if filter.Status != "" {
where += " AND status = ?"
args = append(args, string(filter.Status))
}
args = append(args, filter.Limit, filter.Offset)
rows, err := s.db.QueryContext(ctx, `
SELECT id, chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at
FROM tasks WHERE `+where+` ORDER BY updated_at DESC LIMIT ? OFFSET ?`, args...)
if err != nil {
return nil, fmt.Errorf("%w: list tasks: %w", ErrDB, err)
}
defer rows.Close()
var tasks []*Task
for rows.Next() {
t := &Task{}
if err := rows.Scan(&t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo,
&t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt); err != nil {
return nil, fmt.Errorf("%w: scan task: %w", ErrDB, err)
}
tasks = append(tasks, t)
}
return tasks, rows.Err()
}
// DeleteTask удаляет задачу (каскадно — трассы). Только для тестов/админки.
func (s *Storage) DeleteTask(ctx context.Context, id int64) error {
res, err := s.db.ExecContext(ctx, `DELETE FROM tasks WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("%w: delete task %d: %w", ErrDB, id, err)
}
n, _ := res.RowsAffected()
if n == 0 {
return fmt.Errorf("%w: task %d", ErrNotFound, id)
}
return nil
}

112
internal/storage/traces.go Normal file
View File

@@ -0,0 +1,112 @@
package storage
import (
"context"
"database/sql"
"fmt"
)
// AppendTrace добавляет трассу выполнения к задаче. Возвращает её ID.
func (s *Storage) AppendTrace(ctx context.Context, tr *Trace) (int64, error) {
now := Now()
res, err := s.db.ExecContext(ctx, `
INSERT INTO traces (task_id, agent, session_id, prompt, output, status, started_at)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
tr.TaskID, tr.Agent, tr.SessionID, tr.Prompt, tr.Output, TraceRunning, now,
)
if err != nil {
return 0, fmt.Errorf("%w: append trace: %w", ErrDB, err)
}
id, err := res.LastInsertId()
if err != nil {
return 0, fmt.Errorf("%w: last insert id: %w", ErrDB, err)
}
tr.ID = id
tr.Status = TraceRunning
tr.StartedAt = now
return id, nil
}
// UpdateTraceStatus обновляет статус и finished_at трассы.
func (s *Storage) UpdateTraceStatus(ctx context.Context, traceID int64, status TraceStatus) error {
now := Now()
res, err := s.db.ExecContext(ctx, `
UPDATE traces SET status=?, finished_at=? WHERE id=?`,
status, now, traceID,
)
if err != nil {
return fmt.Errorf("%w: update trace %d: %w", ErrDB, traceID, err)
}
n, _ := res.RowsAffected()
if n == 0 {
return fmt.Errorf("%w: trace %d", ErrTraceNotFound, traceID)
}
return nil
}
// UpdateTraceOutput обновляет output трассы (например, по мере поступления данных).
func (s *Storage) UpdateTraceOutput(ctx context.Context, traceID int64, output string) error {
_, err := s.db.ExecContext(ctx, `UPDATE traces SET output=? WHERE id=?`, output, traceID)
if err != nil {
return fmt.Errorf("%w: update trace output %d: %w", ErrDB, traceID, err)
}
return nil
}
// GetTraces возвращает трассы для задачи, отсортированные по started_at.
func (s *Storage) GetTraces(ctx context.Context, taskID int64) ([]*Trace, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT id, task_id, agent, session_id, prompt, output, status, started_at, finished_at
FROM traces WHERE task_id=? ORDER BY started_at ASC`, taskID)
if err != nil {
return nil, fmt.Errorf("%w: get traces for task %d: %w", ErrDB, taskID, err)
}
defer rows.Close()
var traces []*Trace
for rows.Next() {
tr := &Trace{}
var finishedAt NullSQLiteTime
if err := rows.Scan(&tr.ID, &tr.TaskID, &tr.Agent, &tr.SessionID,
&tr.Prompt, &tr.Output, &tr.Status, &tr.StartedAt, &finishedAt); err != nil {
return nil, fmt.Errorf("%w: scan trace: %w", ErrDB, err)
}
tr.FinishedAt = finishedAt
traces = append(traces, tr)
}
return traces, rows.Err()
}
// GetLatestTrace возвращает последнюю трассу задачи для заданного агента.
func (s *Storage) GetLatestTrace(ctx context.Context, taskID int64, agent string) (*Trace, error) {
tr := &Trace{}
var finishedAt NullSQLiteTime
err := s.db.QueryRowContext(ctx, `
SELECT id, task_id, agent, session_id, prompt, output, status, started_at, finished_at
FROM traces WHERE task_id=? AND agent=? ORDER BY id DESC LIMIT 1`,
taskID, agent).Scan(
&tr.ID, &tr.TaskID, &tr.Agent, &tr.SessionID,
&tr.Prompt, &tr.Output, &tr.Status, &tr.StartedAt, &finishedAt,
)
if err == sql.ErrNoRows {
return nil, fmt.Errorf("%w: trace for task %d agent %s", ErrTraceNotFound, taskID, agent)
}
if err != nil {
return nil, fmt.Errorf("%w: get latest trace: %w", ErrDB, err)
}
tr.FinishedAt = finishedAt
return tr, nil
}
// DeleteTrace удаляет трассу. Только для тестов/админки.
func (s *Storage) DeleteTrace(ctx context.Context, id int64) error {
res, err := s.db.ExecContext(ctx, `DELETE FROM traces WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("%w: delete trace %d: %w", ErrDB, id, err)
}
n, _ := res.RowsAffected()
if n == 0 {
return fmt.Errorf("%w: trace %d", ErrTraceNotFound, id)
}
return nil
}