Refactor db pkg
This commit is contained in:
+54
-13
@@ -2,23 +2,64 @@ package db
|
||||
|
||||
import "time"
|
||||
|
||||
// ApprovalStatus defines the lifecycle states for a tool call approval.
|
||||
type ApprovalStatus string
|
||||
// ExecutionStatus is the state of one tool call, as stored in the executions
|
||||
// table (one row for every tool call in an assistant message).
|
||||
//
|
||||
// The row tracks the call's lifecycle; it does not store the call's
|
||||
// outcome. The outcome lives in a tool result message row in the messages
|
||||
// table, matched to the call by tool_call_id.
|
||||
//
|
||||
// Lifecycle, by tool kind:
|
||||
//
|
||||
// A. Approval not required:
|
||||
//
|
||||
// ┌────────────┐ ┌────────────┐
|
||||
// │ running ├┄┄┄▶│ completed │
|
||||
// └────────────┘ └────────────┘
|
||||
//
|
||||
// B. Approval required:
|
||||
//
|
||||
// ┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐
|
||||
// │ pending ├───▶│ approved ├───▶│ running ├┄┄┄▶│ completed │
|
||||
// └────────────┘ └────────────┘ └────────────┘ └────────────┘
|
||||
// │
|
||||
// ▼
|
||||
// ┌────────────┐
|
||||
// │ denied │
|
||||
// └────────────┘
|
||||
//
|
||||
// The dashed arrow from running to completed indicates the possibility that a
|
||||
// tool call may not complete. if a tool call is interrupted, the row ends at
|
||||
// running. See [ExecutionStatusRunning].
|
||||
type ExecutionStatus string
|
||||
|
||||
const (
|
||||
ApprovalStatusApproved ApprovalStatus = "approved"
|
||||
ApprovalStatusDenied ApprovalStatus = "denied"
|
||||
ApprovalStatusPending ApprovalStatus = "pending"
|
||||
// ExecutionStatusApproved: tool call approved by user.
|
||||
// The call has not launched yet.
|
||||
ExecutionStatusApproved ExecutionStatus = "approved"
|
||||
// ExecutionStatusCompleted: the tool call returned output or an error.
|
||||
// Terminal.
|
||||
ExecutionStatusCompleted ExecutionStatus = "completed"
|
||||
// ExecutionStatusDenied: user refused the tool call. Terminal.
|
||||
ExecutionStatusDenied ExecutionStatus = "denied"
|
||||
// ExecutionStatusPending: the call awaits a user decision.
|
||||
ExecutionStatusPending ExecutionStatus = "pending"
|
||||
// ExecutionStatusRunning: the tool call either:
|
||||
// A. is about to be started,
|
||||
// B. was started and is still executing,
|
||||
// C. was interrupted before it started or finished (terminal).
|
||||
ExecutionStatusRunning ExecutionStatus = "running"
|
||||
)
|
||||
|
||||
// Approval represents a tool call awaiting or resolved by user permission.
|
||||
type Approval struct {
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
Id string `json:"id"`
|
||||
MessageId string `json:"message_id"`
|
||||
Status ApprovalStatus `json:"status"`
|
||||
ToolCallId string `json:"tool_call_id"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
// Execution represents a row in the executions table: one tool call
|
||||
// within an assistant message and its current state.
|
||||
type Execution struct {
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
Id string `json:"id"`
|
||||
MessageId string `json:"message_id"`
|
||||
Status ExecutionStatus `json:"status"`
|
||||
ToolCallId string `json:"tool_call_id"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// Conversation represents a stored conversation.
|
||||
|
||||
+110
-108
@@ -52,9 +52,8 @@ func (q *Queries) InsertUser(ctx context.Context, arg InsertUserParams) (*User,
|
||||
u User
|
||||
createdAt string
|
||||
updatedAt string
|
||||
err error
|
||||
)
|
||||
if err := row.Scan(
|
||||
err := row.Scan(
|
||||
&createdAt,
|
||||
&u.DeactivatedAt,
|
||||
&u.Id,
|
||||
@@ -62,7 +61,8 @@ func (q *Queries) InsertUser(ctx context.Context, arg InsertUserParams) (*User,
|
||||
&u.Password,
|
||||
&u.Role,
|
||||
&updatedAt,
|
||||
); err != nil {
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"InsertUser: %w", err,
|
||||
)
|
||||
@@ -457,208 +457,210 @@ func (q *Queries) NextSequence(ctx context.Context, conversationId string) (int,
|
||||
return seq + 1, nil
|
||||
}
|
||||
|
||||
const insertApproval = `
|
||||
INSERT INTO approvals (id, message_id, tool_call_id)
|
||||
VALUES (?, ?, ?)
|
||||
const insertExecution = `
|
||||
INSERT INTO executions (id, message_id, status, tool_call_id)
|
||||
VALUES (?, ?, ?, ?)
|
||||
RETURNING created_at, id, message_id, status, tool_call_id, updated_at
|
||||
`
|
||||
|
||||
// InsertApprovalParams holds arguments for InsertApproval.
|
||||
type InsertApprovalParams struct {
|
||||
// InsertExecutionParams holds arguments for InsertExecution.
|
||||
type InsertExecutionParams struct {
|
||||
MessageId string
|
||||
Status ExecutionStatus
|
||||
ToolCallId string
|
||||
}
|
||||
|
||||
// InsertApproval creates a new pending approval.
|
||||
func (q *Queries) InsertApproval(ctx context.Context, arg InsertApprovalParams) (*Approval, error) {
|
||||
// InsertExecution creates a new execution row.
|
||||
func (q *Queries) InsertExecution(ctx context.Context, arg InsertExecutionParams) (*Execution, error) {
|
||||
row := q.db.QueryRowContext(
|
||||
ctx,
|
||||
insertApproval,
|
||||
insertExecution,
|
||||
uuid.New().String(),
|
||||
arg.MessageId,
|
||||
arg.Status,
|
||||
arg.ToolCallId,
|
||||
)
|
||||
|
||||
var (
|
||||
a Approval
|
||||
createdAt string
|
||||
updatedAt string
|
||||
err error
|
||||
)
|
||||
if err = row.Scan(
|
||||
&createdAt,
|
||||
&a.Id,
|
||||
&a.MessageId,
|
||||
&a.Status,
|
||||
&a.ToolCallId,
|
||||
&updatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"InsertApproval: %w", err,
|
||||
)
|
||||
}
|
||||
if a.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil {
|
||||
return nil, fmt.Errorf("InsertApproval: parse created_at: %w", err)
|
||||
}
|
||||
if a.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil {
|
||||
return nil, fmt.Errorf("InsertApproval: parse updated_at: %w", err)
|
||||
}
|
||||
return &a, nil
|
||||
}
|
||||
|
||||
const selectApprovalId = `
|
||||
SELECT created_at, id, message_id, status, tool_call_id, updated_at
|
||||
FROM approvals
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
// SelectApprovalId returns a single approval by ID.
|
||||
func (q *Queries) SelectApprovalId(ctx context.Context, id string) (*Approval, error) {
|
||||
row := q.db.QueryRowContext(ctx, selectApprovalId, id)
|
||||
|
||||
var (
|
||||
a Approval
|
||||
e Execution
|
||||
createdAt string
|
||||
updatedAt string
|
||||
)
|
||||
err := row.Scan(
|
||||
&createdAt,
|
||||
&a.Id,
|
||||
&a.MessageId,
|
||||
&a.Status,
|
||||
&a.ToolCallId,
|
||||
&e.Id,
|
||||
&e.MessageId,
|
||||
&e.Status,
|
||||
&e.ToolCallId,
|
||||
&updatedAt,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("SelectApprovalId: %w", err)
|
||||
return nil, fmt.Errorf(
|
||||
"InsertExecution: %w", err,
|
||||
)
|
||||
}
|
||||
if a.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectApprovalId: parse created_at: %w", err)
|
||||
if e.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil {
|
||||
return nil, fmt.Errorf("InsertExecution: parse created_at: %w", err)
|
||||
}
|
||||
if a.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectApprovalId: parse updated_at: %w", err)
|
||||
if e.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil {
|
||||
return nil, fmt.Errorf("InsertExecution: parse updated_at: %w", err)
|
||||
}
|
||||
return &a, nil
|
||||
return &e, nil
|
||||
}
|
||||
|
||||
const selectApprovalsMessageId = `
|
||||
const selectExecutionId = `
|
||||
SELECT created_at, id, message_id, status, tool_call_id, updated_at
|
||||
FROM approvals
|
||||
FROM executions
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
// SelectExecutionId returns a single execution row by ID.
|
||||
func (q *Queries) SelectExecutionId(ctx context.Context, id string) (*Execution, error) {
|
||||
row := q.db.QueryRowContext(ctx, selectExecutionId, id)
|
||||
|
||||
var (
|
||||
e Execution
|
||||
createdAt string
|
||||
updatedAt string
|
||||
)
|
||||
err := row.Scan(
|
||||
&createdAt,
|
||||
&e.Id,
|
||||
&e.MessageId,
|
||||
&e.Status,
|
||||
&e.ToolCallId,
|
||||
&updatedAt,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("SelectExecutionId: %w", err)
|
||||
}
|
||||
if e.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectExecutionId: parse created_at: %w", err)
|
||||
}
|
||||
if e.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectExecutionId: parse updated_at: %w", err)
|
||||
}
|
||||
return &e, nil
|
||||
}
|
||||
|
||||
const selectExecutionsMessageId = `
|
||||
SELECT created_at, id, message_id, status, tool_call_id, updated_at
|
||||
FROM executions
|
||||
WHERE message_id = ?
|
||||
ORDER BY created_at
|
||||
`
|
||||
|
||||
// SelectApprovalsMessageId returns all approvals for a message,
|
||||
// SelectExecutionsMessageId returns all execution rows for a message,
|
||||
// ordered by creation time.
|
||||
func (q *Queries) SelectApprovalsMessageId(ctx context.Context, messageId string) ([]Approval, error) {
|
||||
rows, err := q.db.QueryContext(ctx, selectApprovalsMessageId, messageId)
|
||||
func (q *Queries) SelectExecutionsMessageId(ctx context.Context, messageId string) ([]Execution, error) {
|
||||
rows, err := q.db.QueryContext(ctx, selectExecutionsMessageId, messageId)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("SelectApprovalsMessageId: %w", err)
|
||||
return nil, fmt.Errorf("SelectExecutionsMessageId: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var approvals []Approval
|
||||
var executions []Execution
|
||||
for rows.Next() {
|
||||
var (
|
||||
a Approval
|
||||
e Execution
|
||||
createdAt string
|
||||
updatedAt string
|
||||
)
|
||||
if err := rows.Scan(
|
||||
&createdAt,
|
||||
&a.Id,
|
||||
&a.MessageId,
|
||||
&a.Status,
|
||||
&a.ToolCallId,
|
||||
&e.Id,
|
||||
&e.MessageId,
|
||||
&e.Status,
|
||||
&e.ToolCallId,
|
||||
&updatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("SelectApprovalsMessageId: %w", err)
|
||||
return nil, fmt.Errorf("SelectExecutionsMessageId: %w", err)
|
||||
}
|
||||
if a.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectApprovalsMessageId: parse created_at: %w", err)
|
||||
if e.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectExecutionsMessageId: parse created_at: %w", err)
|
||||
}
|
||||
if a.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectApprovalsMessageId: parse updated_at: %w", err)
|
||||
if e.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectExecutionsMessageId: parse updated_at: %w", err)
|
||||
}
|
||||
approvals = append(approvals, a)
|
||||
executions = append(executions, e)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("SelectApprovalsMessageId: %w", err)
|
||||
return nil, fmt.Errorf("SelectExecutionsMessageId: %w", err)
|
||||
}
|
||||
return approvals, nil
|
||||
return executions, nil
|
||||
}
|
||||
|
||||
const selectPendingApprovals = `
|
||||
const selectPendingExecutions = `
|
||||
SELECT created_at, id, message_id, status, tool_call_id, updated_at
|
||||
FROM approvals
|
||||
FROM executions
|
||||
WHERE status = 'pending'
|
||||
ORDER BY created_at
|
||||
`
|
||||
|
||||
// SelectPendingApprovals returns all pending approvals across all
|
||||
// SelectPendingExecutions returns all pending execution rows across all
|
||||
// conversations, ordered by creation time. This serves as the global
|
||||
// inbox for the UI.
|
||||
func (q *Queries) SelectPendingApprovals(ctx context.Context) ([]Approval, error) {
|
||||
rows, err := q.db.QueryContext(ctx, selectPendingApprovals)
|
||||
func (q *Queries) SelectPendingExecutions(ctx context.Context) ([]Execution, error) {
|
||||
rows, err := q.db.QueryContext(ctx, selectPendingExecutions)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("SelectPendingApprovals: %w", err)
|
||||
return nil, fmt.Errorf("SelectPendingExecutions: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var approvals []Approval
|
||||
var executions []Execution
|
||||
for rows.Next() {
|
||||
var (
|
||||
a Approval
|
||||
e Execution
|
||||
createdAt string
|
||||
updatedAt string
|
||||
)
|
||||
if err := rows.Scan(
|
||||
&createdAt,
|
||||
&a.Id,
|
||||
&a.MessageId,
|
||||
&a.Status,
|
||||
&a.ToolCallId,
|
||||
&e.Id,
|
||||
&e.MessageId,
|
||||
&e.Status,
|
||||
&e.ToolCallId,
|
||||
&updatedAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("SelectPendingApprovals: %w", err)
|
||||
return nil, fmt.Errorf("SelectPendingExecutions: %w", err)
|
||||
}
|
||||
if a.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectPendingApprovals: parse created_at: %w", err)
|
||||
if e.CreatedAt, err = time.Parse(time.RFC3339, createdAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectPendingExecutions: parse created_at: %w", err)
|
||||
}
|
||||
if a.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectPendingApprovals: parse updated_at: %w", err)
|
||||
if e.UpdatedAt, err = time.Parse(time.RFC3339, updatedAt); err != nil {
|
||||
return nil, fmt.Errorf("SelectPendingExecutions: parse updated_at: %w", err)
|
||||
}
|
||||
approvals = append(approvals, a)
|
||||
executions = append(executions, e)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("SelectPendingApprovals: %w", err)
|
||||
return nil, fmt.Errorf("SelectPendingExecutions: %w", err)
|
||||
}
|
||||
return approvals, nil
|
||||
return executions, nil
|
||||
}
|
||||
|
||||
const updateApprovalStatus = `
|
||||
UPDATE approvals
|
||||
const updateExecutionStatus = `
|
||||
UPDATE executions
|
||||
SET status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
|
||||
WHERE id = ?
|
||||
`
|
||||
|
||||
// UpdateApprovalStatusParams holds arguments for UpdateApprovalStatus.
|
||||
type UpdateApprovalStatusParams struct {
|
||||
// UpdateExecutionStatusParams holds arguments for UpdateExecutionStatus.
|
||||
type UpdateExecutionStatusParams struct {
|
||||
Id string
|
||||
Status ApprovalStatus
|
||||
Status ExecutionStatus
|
||||
}
|
||||
|
||||
// UpdateApprovalStatus transitions an approval's status and bumps
|
||||
// UpdateExecutionStatus transitions an execution row's status and bumps
|
||||
// updated_at. The caller is responsible for valid transitions.
|
||||
func (q *Queries) UpdateApprovalStatus(ctx context.Context, arg UpdateApprovalStatusParams) error {
|
||||
func (q *Queries) UpdateExecutionStatus(ctx context.Context, arg UpdateExecutionStatusParams) error {
|
||||
_, err := q.db.ExecContext(
|
||||
ctx,
|
||||
updateApprovalStatus,
|
||||
updateExecutionStatus,
|
||||
arg.Status,
|
||||
arg.Id,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("UpdateApprovalStatus: %w", err)
|
||||
return fmt.Errorf("UpdateExecutionStatus: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
+166
-47
@@ -2,8 +2,11 @@ package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"testing"
|
||||
|
||||
"github.com/mattn/go-sqlite3"
|
||||
)
|
||||
|
||||
func TestQueries(t *testing.T) {
|
||||
@@ -29,8 +32,8 @@ func TestQueries(t *testing.T) {
|
||||
t.Run("ConversationTimestamp", func(t *testing.T) {
|
||||
testConversationTimestamp(t, ctx, db)
|
||||
})
|
||||
t.Run("Approvals", func(t *testing.T) {
|
||||
testApprovals(t, ctx, db)
|
||||
t.Run("Executions", func(t *testing.T) {
|
||||
testExecutions(t, ctx, db)
|
||||
})
|
||||
t.Run("Delete", func(t *testing.T) {
|
||||
testDelete(t, ctx, db)
|
||||
@@ -380,7 +383,7 @@ func testDelete(t *testing.T, ctx context.Context, db *SQLite) {
|
||||
}
|
||||
}
|
||||
|
||||
func testApprovals(t *testing.T, ctx context.Context, db *SQLite) {
|
||||
func testExecutions(t *testing.T, ctx context.Context, db *SQLite) {
|
||||
var (
|
||||
qr = New(db.Reader)
|
||||
qw = New(db.Writer)
|
||||
@@ -412,112 +415,228 @@ func testApprovals(t *testing.T, ctx context.Context, db *SQLite) {
|
||||
t.Fatalf("InsertMessage: %v", err)
|
||||
}
|
||||
|
||||
// Insert an approval.
|
||||
approval, err := qw.InsertApproval(ctx, InsertApprovalParams{
|
||||
// Insert an execution.
|
||||
execution, err := qw.InsertExecution(ctx, InsertExecutionParams{
|
||||
MessageId: msg.Id,
|
||||
Status: ExecutionStatusPending,
|
||||
ToolCallId: "call_abc",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("InsertApproval: %v", err)
|
||||
t.Fatalf("InsertExecution: %v", err)
|
||||
}
|
||||
if approval.Status != ApprovalStatusPending {
|
||||
t.Fatalf("expected pending, got %s", approval.Status)
|
||||
if execution.Status != ExecutionStatusPending {
|
||||
t.Fatalf("expected pending, got %s", execution.Status)
|
||||
}
|
||||
if approval.ToolCallId != "call_abc" {
|
||||
t.Fatalf("expected tool_call_id call_abc, got %s", approval.ToolCallId)
|
||||
if execution.ToolCallId != "call_abc" {
|
||||
t.Fatalf("expected tool_call_id call_abc, got %s", execution.ToolCallId)
|
||||
}
|
||||
|
||||
// Select by ID.
|
||||
res, err := qr.SelectApprovalId(ctx, approval.Id)
|
||||
res, err := qr.SelectExecutionId(ctx, execution.Id)
|
||||
if err != nil {
|
||||
t.Fatalf("SelectApprovalId: %v", err)
|
||||
t.Fatalf("SelectExecutionId: %v", err)
|
||||
}
|
||||
if res == nil {
|
||||
t.Fatal("SelectApprovalId returned nil")
|
||||
t.Fatal("SelectExecutionId returned nil")
|
||||
}
|
||||
if res.MessageId != msg.Id {
|
||||
t.Errorf("expected message_id %s, got %s", msg.Id, res.MessageId)
|
||||
}
|
||||
|
||||
// Select by message ID.
|
||||
approvals, err := qr.SelectApprovalsMessageId(ctx, msg.Id)
|
||||
executions, err := qr.SelectExecutionsMessageId(ctx, msg.Id)
|
||||
if err != nil {
|
||||
t.Fatalf("SelectApprovalsMessageId: %v", err)
|
||||
t.Fatalf("SelectExecutionsMessageId: %v", err)
|
||||
}
|
||||
if len(approvals) != 1 {
|
||||
t.Fatalf("expected 1 approval, got %d", len(approvals))
|
||||
if len(executions) != 1 {
|
||||
t.Fatalf("expected 1 execution, got %d", len(executions))
|
||||
}
|
||||
|
||||
// Select pending.
|
||||
pending, err := qr.SelectPendingApprovals(ctx)
|
||||
pending, err := qr.SelectPendingExecutions(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("SelectPendingApprovals: %v", err)
|
||||
t.Fatalf("SelectPendingExecutions: %v", err)
|
||||
}
|
||||
if len(pending) != 1 {
|
||||
t.Fatalf("expected 1 pending approval, got %d", len(pending))
|
||||
t.Fatalf("expected 1 pending execution, got %d", len(pending))
|
||||
}
|
||||
|
||||
// Transition to approved.
|
||||
if err := qw.UpdateApprovalStatus(ctx, UpdateApprovalStatusParams{
|
||||
Id: approval.Id,
|
||||
Status: ApprovalStatusApproved,
|
||||
if err := qw.UpdateExecutionStatus(ctx, UpdateExecutionStatusParams{
|
||||
Id: execution.Id,
|
||||
Status: ExecutionStatusApproved,
|
||||
}); err != nil {
|
||||
t.Fatalf("UpdateApprovalStatus: %v", err)
|
||||
t.Fatalf("UpdateExecutionStatus: %v", err)
|
||||
}
|
||||
|
||||
res, err = qr.SelectApprovalId(ctx, approval.Id)
|
||||
res, err = qr.SelectExecutionId(ctx, execution.Id)
|
||||
if err != nil {
|
||||
t.Fatalf("SelectApprovalId after update: %v", err)
|
||||
t.Fatalf("SelectExecutionId after update: %v", err)
|
||||
}
|
||||
if res.Status != ApprovalStatusApproved {
|
||||
if res.Status != ExecutionStatusApproved {
|
||||
t.Fatalf("expected approved, got %s", res.Status)
|
||||
}
|
||||
|
||||
// Pending should now be empty.
|
||||
pending, err = qr.SelectPendingApprovals(ctx)
|
||||
pending, err = qr.SelectPendingExecutions(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("SelectPendingApprovals after update: %v", err)
|
||||
t.Fatalf("SelectPendingExecutions after update: %v", err)
|
||||
}
|
||||
if len(pending) != 0 {
|
||||
t.Fatalf("expected 0 pending, got %d", len(pending))
|
||||
}
|
||||
|
||||
// Insert a second approval and deny it.
|
||||
approval2, err := qw.InsertApproval(ctx, InsertApprovalParams{
|
||||
// Insert a second execution and deny it.
|
||||
execution2, err := qw.InsertExecution(ctx, InsertExecutionParams{
|
||||
MessageId: msg.Id,
|
||||
Status: ExecutionStatusPending,
|
||||
ToolCallId: "call_def",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("InsertApproval second: %v", err)
|
||||
t.Fatalf("InsertExecution second: %v", err)
|
||||
}
|
||||
|
||||
if err := qw.UpdateApprovalStatus(ctx, UpdateApprovalStatusParams{
|
||||
Id: approval2.Id,
|
||||
Status: ApprovalStatusDenied,
|
||||
if err := qw.UpdateExecutionStatus(ctx, UpdateExecutionStatusParams{
|
||||
Id: execution2.Id,
|
||||
Status: ExecutionStatusDenied,
|
||||
}); err != nil {
|
||||
t.Fatalf("UpdateApprovalStatus denied: %v", err)
|
||||
t.Fatalf("UpdateExecutionStatus denied: %v", err)
|
||||
}
|
||||
|
||||
res2, err := qr.SelectApprovalId(ctx, approval2.Id)
|
||||
res2, err := qr.SelectExecutionId(ctx, execution2.Id)
|
||||
if err != nil {
|
||||
t.Fatalf("SelectApprovalId after deny: %v", err)
|
||||
t.Fatalf("SelectExecutionId after deny: %v", err)
|
||||
}
|
||||
if res2.Status != ApprovalStatusDenied {
|
||||
if res2.Status != ExecutionStatusDenied {
|
||||
t.Fatalf("expected denied, got %s", res2.Status)
|
||||
}
|
||||
|
||||
// Both approvals for the message, ordered by created_at.
|
||||
all, err := qr.SelectApprovalsMessageId(ctx, msg.Id)
|
||||
// Transition to running, then completed.
|
||||
if err := qw.UpdateExecutionStatus(ctx, UpdateExecutionStatusParams{
|
||||
Id: execution.Id,
|
||||
Status: ExecutionStatusRunning,
|
||||
}); err != nil {
|
||||
t.Fatalf("UpdateExecutionStatus running: %v", err)
|
||||
}
|
||||
if err := qw.UpdateExecutionStatus(ctx, UpdateExecutionStatusParams{
|
||||
Id: execution.Id,
|
||||
Status: ExecutionStatusCompleted,
|
||||
}); err != nil {
|
||||
t.Fatalf("UpdateExecutionStatus completed: %v", err)
|
||||
}
|
||||
|
||||
// Both executions for the message, ordered by created_at.
|
||||
all, err := qr.SelectExecutionsMessageId(ctx, msg.Id)
|
||||
if err != nil {
|
||||
t.Fatalf("SelectApprovalsMessageId final: %v", err)
|
||||
t.Fatalf("SelectExecutionsMessageId final: %v", err)
|
||||
}
|
||||
if len(all) != 2 {
|
||||
t.Fatalf("expected 2 approvals, got %d", len(all))
|
||||
t.Fatalf("expected 2 executions, got %d", len(all))
|
||||
}
|
||||
if all[0].Status != ApprovalStatusApproved {
|
||||
t.Fatalf("expected first approval approved, got %s", all[0].Status)
|
||||
if all[0].Status != ExecutionStatusCompleted {
|
||||
t.Fatalf("expected first execution completed, got %s", all[0].Status)
|
||||
}
|
||||
if all[1].Status != ApprovalStatusDenied {
|
||||
t.Fatalf("expected second approval denied, got %s", all[1].Status)
|
||||
if all[1].Status != ExecutionStatusDenied {
|
||||
t.Fatalf("expected second execution denied, got %s", all[1].Status)
|
||||
}
|
||||
|
||||
// Parameter values and the schema default, on a second message.
|
||||
msg2, err := qw.InsertMessage(ctx, InsertMessageParams{
|
||||
ConversationId: c.Id,
|
||||
Sequence: 2,
|
||||
Role: MessageRoleAssistant,
|
||||
Content: `{"role":"assistant","content":"second"}`,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("InsertMessage second: %v", err)
|
||||
}
|
||||
|
||||
execRunning, err := qw.InsertExecution(ctx, InsertExecutionParams{
|
||||
MessageId: msg2.Id,
|
||||
Status: ExecutionStatusRunning,
|
||||
ToolCallId: "call_running",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("InsertExecution running: %v", err)
|
||||
}
|
||||
if execRunning.Status != ExecutionStatusRunning {
|
||||
t.Fatalf("expected running, got %s", execRunning.Status)
|
||||
}
|
||||
|
||||
// A status other than running is stored as passed.
|
||||
execPending, err := qw.InsertExecution(ctx, InsertExecutionParams{
|
||||
MessageId: msg2.Id,
|
||||
Status: ExecutionStatusPending,
|
||||
ToolCallId: "call_pending",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("InsertExecution pending: %v", err)
|
||||
}
|
||||
if execPending.Status != ExecutionStatusPending {
|
||||
t.Fatalf("expected pending, got %s", execPending.Status)
|
||||
}
|
||||
|
||||
// The zero value is rejected by the table's status check.
|
||||
_, err = qw.InsertExecution(ctx, InsertExecutionParams{
|
||||
MessageId: msg2.Id,
|
||||
ToolCallId: "call_empty",
|
||||
})
|
||||
var sqliteErr sqlite3.Error
|
||||
if !errors.As(err, &sqliteErr) {
|
||||
t.Fatalf("InsertExecution with zero status: expected sqlite3.Error, got %v", err)
|
||||
}
|
||||
if sqliteErr.ExtendedCode != sqlite3.ErrConstraintCheck {
|
||||
t.Fatalf("InsertExecution with zero status: expected ErrConstraintCheck, got %v", err)
|
||||
}
|
||||
|
||||
// A second row for the same message and tool call is rejected.
|
||||
_, err = qw.InsertExecution(ctx, InsertExecutionParams{
|
||||
MessageId: msg2.Id,
|
||||
Status: ExecutionStatusPending,
|
||||
ToolCallId: "call_running",
|
||||
})
|
||||
if !errors.As(err, &sqliteErr) {
|
||||
t.Fatalf("InsertExecution duplicate: expected sqlite3.Error, got %v", err)
|
||||
}
|
||||
if sqliteErr.ExtendedCode != sqlite3.ErrConstraintUnique {
|
||||
t.Fatalf("InsertExecution duplicate: expected ErrConstraintUnique, got %v", err)
|
||||
}
|
||||
|
||||
// A row inserted without a status takes the schema default, running.
|
||||
if _, err := db.Writer.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO executions (id, message_id, tool_call_id) VALUES (?, ?, ?)`,
|
||||
"exec_default", msg2.Id, "call_default",
|
||||
); err != nil {
|
||||
t.Fatalf("insert execution without status: %v", err)
|
||||
}
|
||||
res, err = qr.SelectExecutionId(ctx, "exec_default")
|
||||
if err != nil {
|
||||
t.Fatalf("SelectExecutionId default: %v", err)
|
||||
}
|
||||
if res == nil {
|
||||
t.Fatal("SelectExecutionId default returned nil")
|
||||
}
|
||||
if res.Status != ExecutionStatusRunning {
|
||||
t.Fatalf("expected default running, got %s", res.Status)
|
||||
}
|
||||
|
||||
// The rejected insert stored no row: the message's rows are exactly the
|
||||
// three successful inserts.
|
||||
all2, err := qr.SelectExecutionsMessageId(ctx, msg2.Id)
|
||||
if err != nil {
|
||||
t.Fatalf("SelectExecutionsMessageId second: %v", err)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, e := range all2 {
|
||||
seen[e.ToolCallId] = true
|
||||
}
|
||||
for _, id := range []string{"call_default", "call_pending", "call_running"} {
|
||||
if !seen[id] {
|
||||
t.Fatalf("execution %s not found", id)
|
||||
}
|
||||
}
|
||||
if len(seen) != 3 {
|
||||
t.Fatalf("expected 3 executions, got %d", len(seen))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -99,31 +99,56 @@ BEGIN
|
||||
WHERE id = OLD.conversation_id;
|
||||
END;
|
||||
|
||||
-- Approvals track tool calls that require user permission before
|
||||
-- execution. When the LLM requests a tool call that needs approval, a row
|
||||
-- is inserted with status "pending" and the agent loop pauses. The UI
|
||||
-- queries pending approvals to present an inbox to the user. When the
|
||||
-- user decides, the status transitions to "approved" (proceed) or
|
||||
-- "denied" (refuse), and the agent loop resumes.
|
||||
-- Executions track the state of tool calls. Each row references a
|
||||
-- specific tool call within an assistant message via message_id and
|
||||
-- tool_call_id. The tool call's name and arguments are stored in the
|
||||
-- message's content JSON; they are not duplicated here. The tool's
|
||||
-- execution result (stdout, stderr, exit code) is stored as a
|
||||
-- separate tool-role message in the messages table, not in this row.
|
||||
--
|
||||
-- Each approval references a specific tool call within an assistant
|
||||
-- message via message_id and tool_call_id. The tool call's name and
|
||||
-- arguments are stored in the message's content JSON; they are not
|
||||
-- duplicated here. The tool's execution result (stdout, stderr, exit
|
||||
-- code) is stored as a separate tool-role message in the messages
|
||||
-- table, not in the approval.
|
||||
CREATE TABLE IF NOT EXISTS approvals (
|
||||
-- Lifecycle, by tool kind:
|
||||
--
|
||||
-- A. Approval not required:
|
||||
--
|
||||
-- ┌────────────┐ ┌────────────┐
|
||||
-- │ running ├┄┄┄▶│ completed │
|
||||
-- └────────────┘ └────────────┘
|
||||
--
|
||||
-- B. Approval required:
|
||||
--
|
||||
-- ┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐
|
||||
-- │ pending ├───▶│ approved ├───▶│ running ├┄┄┄▶│ completed │
|
||||
-- └────────────┘ └────────────┘ └────────────┘ └────────────┘
|
||||
-- │
|
||||
-- ▼
|
||||
-- ┌────────────┐
|
||||
-- │ denied │
|
||||
-- └────────────┘
|
||||
--
|
||||
-- The dashed arrow from running to completed indicates the possibility that a
|
||||
-- tool call may not complete. if a tool call is interrupted, the row ends at
|
||||
-- running.
|
||||
--
|
||||
-- approved: tool call approved by user. The call has not launched yet.
|
||||
-- completed: the tool call returned output or an error. Terminal.
|
||||
-- denied: user refused the tool call. Terminal.
|
||||
-- pending: the call awaits a user decision.
|
||||
-- running: the tool call either:
|
||||
-- A. is about to be started,
|
||||
-- B. was started and is still executing,
|
||||
-- C. was interrupted before it started or finished (terminal).
|
||||
CREATE TABLE IF NOT EXISTS executions (
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
|
||||
id TEXT PRIMARY KEY,
|
||||
message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
|
||||
status TEXT NOT NULL DEFAULT 'pending',
|
||||
status TEXT NOT NULL DEFAULT 'running',
|
||||
tool_call_id TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
|
||||
CHECK (status IN ('pending', 'approved', 'denied'))
|
||||
CHECK (status IN ('approved', 'completed', 'denied', 'pending', 'running'))
|
||||
) STRICT;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_approvals_message ON approvals(message_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_approvals_pending ON approvals(status) WHERE status = 'pending';
|
||||
CREATE INDEX IF NOT EXISTS idx_executions_message ON executions(message_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_executions_pending ON executions(status) WHERE status = 'pending';
|
||||
|
||||
-- Default user. Temporary workaround until authentication is built.
|
||||
INSERT INTO users (id, name, password, role)
|
||||
|
||||
@@ -69,8 +69,8 @@ func testApplySchema(t *testing.T, ctx context.Context, db *SQLite) {
|
||||
|
||||
func testTablesExist(t *testing.T, ctx context.Context, db *SQLite) {
|
||||
expected := []string{
|
||||
"approvals",
|
||||
"conversations",
|
||||
"executions",
|
||||
"messages",
|
||||
"sessions",
|
||||
"users",
|
||||
@@ -112,9 +112,9 @@ func testSchemaIdempotent(t *testing.T, ctx context.Context, db *SQLite) {
|
||||
|
||||
func testSchemaIndexes(t *testing.T, ctx context.Context, db *SQLite) {
|
||||
expected := []string{
|
||||
"idx_approvals_message",
|
||||
"idx_approvals_pending",
|
||||
"idx_conv_user",
|
||||
"idx_executions_message",
|
||||
"idx_executions_pending",
|
||||
"idx_msg_conv",
|
||||
"idx_msg_conv_seq",
|
||||
"idx_sessions_active",
|
||||
|
||||
Reference in New Issue
Block a user