Hop
1572 lines
49 KiB
Go
1572 lines
49 KiB
Go
package hop
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
const schemaVersion = 2
|
|
|
|
var (
|
|
ErrNotFound = errors.New("hop: not found")
|
|
ErrNotInitialized = errors.New("hop: repository is not initialized")
|
|
ErrAlreadyInitialized = errors.New("hop: repository is already initialized")
|
|
ErrAttemptHeadChanged = errors.New("hop: attempt head changed")
|
|
ErrAcceptedHeadChanged = errors.New("hop: accepted head changed")
|
|
)
|
|
|
|
// HeadChangedError reports a failed compare-and-swap of an attempt or the
|
|
// repository's accepted head. Callers may use errors.Is with
|
|
// ErrAttemptHeadChanged or ErrAcceptedHeadChanged.
|
|
type HeadChangedError struct {
|
|
Scope string `json:"scope"`
|
|
Expected string `json:"expected"`
|
|
Actual string `json:"actual"`
|
|
}
|
|
|
|
func (e *HeadChangedError) Error() string {
|
|
return fmt.Sprintf("hop: %s head changed: expected %q, found %q", e.Scope, e.Expected, e.Actual)
|
|
}
|
|
|
|
func (e *HeadChangedError) Unwrap() error {
|
|
if e.Scope == "accepted" {
|
|
return ErrAcceptedHeadChanged
|
|
}
|
|
return ErrAttemptHeadChanged
|
|
}
|
|
|
|
// Event is an immutable audit record. Payload is deliberately unstructured so
|
|
// newer clients can add detail without requiring a database migration.
|
|
type Event struct {
|
|
ID string `json:"id"`
|
|
Kind string `json:"kind"`
|
|
TaskID string `json:"task_id,omitempty"`
|
|
AttemptID string `json:"attempt_id,omitempty"`
|
|
StateID string `json:"state_id,omitempty"`
|
|
Payload json.RawMessage `json:"payload,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
}
|
|
|
|
// Store owns Hop's durable state graph. One database connection is used per
|
|
// Store so connection-scoped SQLite settings (notably foreign_keys) cannot be
|
|
// lost when database/sql selects another connection.
|
|
type Store struct {
|
|
db *sql.DB
|
|
path string
|
|
}
|
|
|
|
// OpenStore opens (and, when needed, creates) a Hop SQLite database, applies
|
|
// migrations, and enables foreign keys and WAL mode.
|
|
func OpenStore(path string) (*Store, error) {
|
|
if strings.TrimSpace(path) == "" {
|
|
return nil, errors.New("hop: database path is required")
|
|
}
|
|
if path != ":memory:" && !strings.HasPrefix(path, "file:") {
|
|
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
|
return nil, fmt.Errorf("create database directory: %w", err)
|
|
}
|
|
}
|
|
|
|
db, err := sql.Open("sqlite", path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open database: %w", err)
|
|
}
|
|
db.SetMaxOpenConns(1)
|
|
db.SetMaxIdleConns(1)
|
|
|
|
store := &Store{db: db, path: path}
|
|
if err := store.configure(context.Background()); err != nil {
|
|
_ = db.Close()
|
|
return nil, err
|
|
}
|
|
if err := store.migrate(context.Background()); err != nil {
|
|
_ = db.Close()
|
|
return nil, err
|
|
}
|
|
return store, nil
|
|
}
|
|
|
|
// Open is a convenience alias for OpenStore.
|
|
func Open(path string) (*Store, error) { return OpenStore(path) }
|
|
|
|
func (s *Store) Path() string { return s.path }
|
|
|
|
func (s *Store) Close() error {
|
|
if s == nil || s.db == nil {
|
|
return nil
|
|
}
|
|
return s.db.Close()
|
|
}
|
|
|
|
func (s *Store) configure(ctx context.Context) error {
|
|
for _, statement := range []string{
|
|
"PRAGMA busy_timeout = 5000",
|
|
"PRAGMA foreign_keys = ON",
|
|
"PRAGMA synchronous = NORMAL",
|
|
} {
|
|
if _, err := s.db.ExecContext(ctx, statement); err != nil {
|
|
return fmt.Errorf("configure sqlite (%s): %w", statement, err)
|
|
}
|
|
}
|
|
var journalMode string
|
|
if err := s.db.QueryRowContext(ctx, "PRAGMA journal_mode = WAL").Scan(&journalMode); err != nil {
|
|
return fmt.Errorf("enable sqlite WAL: %w", err)
|
|
}
|
|
var foreignKeys int
|
|
if err := s.db.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&foreignKeys); err != nil {
|
|
return fmt.Errorf("verify sqlite foreign keys: %w", err)
|
|
}
|
|
if foreignKeys != 1 {
|
|
return errors.New("hop: sqlite foreign keys could not be enabled")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type migration struct {
|
|
version int
|
|
statements []string
|
|
}
|
|
|
|
var migrations = []migration{
|
|
{
|
|
version: 1,
|
|
statements: []string{
|
|
`CREATE TABLE IF NOT EXISTS meta (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
) STRICT`,
|
|
`CREATE TABLE IF NOT EXISTS tasks (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
) STRICT`,
|
|
`CREATE TABLE IF NOT EXISTS states (
|
|
id TEXT PRIMARY KEY,
|
|
kind TEXT NOT NULL,
|
|
task_id TEXT REFERENCES tasks(id) ON DELETE SET NULL DEFERRABLE INITIALLY DEFERRED,
|
|
attempt_id TEXT REFERENCES attempts(id) ON DELETE SET NULL DEFERRABLE INITIALLY DEFERRED,
|
|
canonical_anchor_id TEXT REFERENCES states(id) ON DELETE SET NULL DEFERRABLE INITIALLY DEFERRED,
|
|
source_tree TEXT NOT NULL,
|
|
git_commit TEXT NOT NULL,
|
|
prompt TEXT NOT NULL,
|
|
summary TEXT NOT NULL,
|
|
agent TEXT NOT NULL,
|
|
digest TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
) STRICT`,
|
|
`CREATE TABLE IF NOT EXISTS attempts (
|
|
id TEXT PRIMARY KEY,
|
|
task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
|
|
agent TEXT NOT NULL,
|
|
workspace TEXT NOT NULL,
|
|
base_state_id TEXT NOT NULL REFERENCES states(id) ON DELETE RESTRICT DEFERRABLE INITIALLY DEFERRED,
|
|
head_state_id TEXT REFERENCES states(id) ON DELETE SET NULL DEFERRABLE INITIALLY DEFERRED,
|
|
status TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
) STRICT`,
|
|
`CREATE TABLE IF NOT EXISTS state_parents (
|
|
state_id TEXT NOT NULL REFERENCES states(id) ON DELETE CASCADE,
|
|
parent_state_id TEXT NOT NULL REFERENCES states(id) ON DELETE RESTRICT,
|
|
role TEXT NOT NULL,
|
|
parent_order INTEGER NOT NULL,
|
|
PRIMARY KEY (state_id, parent_order),
|
|
UNIQUE (state_id, parent_state_id, role)
|
|
) STRICT`,
|
|
`CREATE TABLE IF NOT EXISTS checks (
|
|
id TEXT PRIMARY KEY,
|
|
attempt_id TEXT NOT NULL REFERENCES attempts(id) ON DELETE CASCADE,
|
|
state_id TEXT REFERENCES states(id) ON DELETE SET NULL,
|
|
tree_hash TEXT NOT NULL,
|
|
command_json TEXT NOT NULL,
|
|
exit_code INTEGER NOT NULL,
|
|
output TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
) STRICT`,
|
|
`CREATE TABLE IF NOT EXISTS events (
|
|
id TEXT PRIMARY KEY,
|
|
kind TEXT NOT NULL,
|
|
task_id TEXT REFERENCES tasks(id) ON DELETE SET NULL,
|
|
attempt_id TEXT REFERENCES attempts(id) ON DELETE SET NULL,
|
|
state_id TEXT REFERENCES states(id) ON DELETE SET NULL,
|
|
payload_json TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
) STRICT`,
|
|
`CREATE INDEX IF NOT EXISTS states_task_created_idx ON states(task_id, created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS states_attempt_created_idx ON states(attempt_id, created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS states_kind_created_idx ON states(kind, created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS state_parents_parent_idx ON state_parents(parent_state_id, state_id)`,
|
|
`CREATE INDEX IF NOT EXISTS state_parents_role_idx ON state_parents(state_id, role, parent_order)`,
|
|
`CREATE INDEX IF NOT EXISTS attempts_task_created_idx ON attempts(task_id, created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS attempts_status_created_idx ON attempts(status, created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS checks_attempt_tree_idx ON checks(attempt_id, tree_hash, created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS checks_tree_idx ON checks(tree_hash, created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS events_created_idx ON events(created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS events_task_idx ON events(task_id, created_at, id)`,
|
|
`CREATE INDEX IF NOT EXISTS events_attempt_idx ON events(attempt_id, created_at, id)`,
|
|
},
|
|
},
|
|
{
|
|
version: 2,
|
|
statements: []string{
|
|
`CREATE TABLE IF NOT EXISTS agent_sessions (
|
|
agent TEXT NOT NULL,
|
|
session_id TEXT NOT NULL,
|
|
head_state_id TEXT NOT NULL REFERENCES states(id) ON DELETE RESTRICT,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (agent, session_id)
|
|
) STRICT`,
|
|
`CREATE INDEX IF NOT EXISTS agent_sessions_head_idx ON agent_sessions(head_state_id)`,
|
|
},
|
|
},
|
|
}
|
|
|
|
func (s *Store) migrate(ctx context.Context) error {
|
|
var current int
|
|
if err := s.db.QueryRowContext(ctx, "PRAGMA user_version").Scan(¤t); err != nil {
|
|
return fmt.Errorf("read schema version: %w", err)
|
|
}
|
|
if current > schemaVersion {
|
|
return fmt.Errorf("hop: database schema version %d is newer than supported version %d", current, schemaVersion)
|
|
}
|
|
for _, m := range migrations {
|
|
if m.version <= current {
|
|
continue
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("begin schema migration %d: %w", m.version, err)
|
|
}
|
|
ok := false
|
|
defer func() {
|
|
if !ok {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
for _, statement := range m.statements {
|
|
if _, err := tx.ExecContext(ctx, statement); err != nil {
|
|
_ = tx.Rollback()
|
|
return fmt.Errorf("apply schema migration %d: %w", m.version, err)
|
|
}
|
|
}
|
|
if _, err := tx.ExecContext(ctx,
|
|
`INSERT INTO meta(key, value) VALUES ('schema_version', ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value`, strconv.Itoa(m.version)); err != nil {
|
|
_ = tx.Rollback()
|
|
return fmt.Errorf("record schema migration %d: %w", m.version, err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, fmt.Sprintf("PRAGMA user_version = %d", m.version)); err != nil {
|
|
_ = tx.Rollback()
|
|
return fmt.Errorf("record sqlite schema version %d: %w", m.version, err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit schema migration %d: %w", m.version, err)
|
|
}
|
|
ok = true
|
|
current = m.version
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// CreateInitialState atomically records the first accepted state and repository
|
|
// root, and makes that state the accepted head.
|
|
func (s *Store) CreateInitialState(ctx context.Context, root string, initial State) (State, error) {
|
|
if strings.TrimSpace(root) == "" {
|
|
return State{}, errors.New("hop: repository root is required")
|
|
}
|
|
if initial.Kind == "" {
|
|
initial.Kind = StateAccepted
|
|
}
|
|
if initial.Kind != StateAccepted {
|
|
return State{}, fmt.Errorf("hop: initial state must have kind %q", StateAccepted)
|
|
}
|
|
if initial.ID == "" {
|
|
initial.ID = newID(stateIDPrefix(initial.Kind))
|
|
}
|
|
if initial.CreatedAt.IsZero() {
|
|
initial.CreatedAt = time.Now().UTC()
|
|
}
|
|
initial.TaskID = ""
|
|
initial.AttemptID = ""
|
|
initial.CanonicalAnchorID = ""
|
|
initial.Parents = nil
|
|
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("begin repository initialization: %w", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
var existing string
|
|
err = tx.QueryRowContext(ctx, `SELECT value FROM meta WHERE key = 'accepted_head'`).Scan(&existing)
|
|
if err == nil {
|
|
return State{}, ErrAlreadyInitialized
|
|
}
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return State{}, fmt.Errorf("check repository initialization: %w", err)
|
|
}
|
|
|
|
if err := insertStateTx(ctx, tx, initial, nil); err != nil {
|
|
return State{}, err
|
|
}
|
|
for key, value := range map[string]string{"root": filepath.Clean(root), "accepted_head": initial.ID} {
|
|
if _, err := tx.ExecContext(ctx,
|
|
`INSERT INTO meta(key, value) VALUES (?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value`, key, value); err != nil {
|
|
return State{}, fmt.Errorf("write repository metadata %q: %w", key, err)
|
|
}
|
|
}
|
|
if err := insertEventTx(ctx, tx, Event{
|
|
Kind: "repository.initialized",
|
|
StateID: initial.ID,
|
|
}, map[string]string{"root": filepath.Clean(root)}); err != nil {
|
|
return State{}, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return State{}, fmt.Errorf("commit repository initialization: %w", err)
|
|
}
|
|
return initial, nil
|
|
}
|
|
|
|
// Initialize is a convenience alias for CreateInitialState.
|
|
func (s *Store) Initialize(ctx context.Context, root string, initial State) (State, error) {
|
|
return s.CreateInitialState(ctx, root, initial)
|
|
}
|
|
|
|
// IsInitialized reports whether an accepted head has been installed.
|
|
func (s *Store) IsInitialized(ctx context.Context) (bool, error) {
|
|
_, err := s.AcceptedHeadID(ctx)
|
|
if errors.Is(err, ErrNotInitialized) {
|
|
return false, nil
|
|
}
|
|
return err == nil, err
|
|
}
|
|
|
|
// AgentSessionHead returns the latest prompt state associated with an agent
|
|
// session. Session heads let interactive agents turn follow-up messages into
|
|
// descendants without asking the human to carry Hop state IDs between turns.
|
|
func (s *Store) AgentSessionHead(ctx context.Context, agent, sessionID string) (string, bool, error) {
|
|
if strings.TrimSpace(agent) == "" || strings.TrimSpace(sessionID) == "" {
|
|
return "", false, errors.New("hop: agent and session ID are required")
|
|
}
|
|
var stateID string
|
|
err := s.db.QueryRowContext(ctx,
|
|
`SELECT head_state_id FROM agent_sessions WHERE agent = ? AND session_id = ?`,
|
|
agent, sessionID).Scan(&stateID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, nil
|
|
}
|
|
if err != nil {
|
|
return "", false, fmt.Errorf("read agent session head: %w", err)
|
|
}
|
|
return stateID, true, nil
|
|
}
|
|
|
|
// SetAgentSessionHead records the prompt state that should parent the next
|
|
// message in an interactive agent session.
|
|
func (s *Store) SetAgentSessionHead(ctx context.Context, agent, sessionID, stateID string) error {
|
|
if strings.TrimSpace(agent) == "" || strings.TrimSpace(sessionID) == "" || strings.TrimSpace(stateID) == "" {
|
|
return errors.New("hop: agent, session ID, and state ID are required")
|
|
}
|
|
if redacted, findings := RedactPromptSecrets(agent + "\n" + sessionID); len(findings) > 0 || redacted != agent+"\n"+sessionID {
|
|
return errors.New("hop: refusing to persist a potential credential in agent session metadata")
|
|
}
|
|
now := formatTime(time.Now().UTC())
|
|
if _, err := s.db.ExecContext(ctx,
|
|
`INSERT INTO agent_sessions(agent, session_id, head_state_id, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
ON CONFLICT(agent, session_id) DO UPDATE SET
|
|
head_state_id = excluded.head_state_id,
|
|
updated_at = excluded.updated_at`,
|
|
agent, sessionID, stateID, now, now); err != nil {
|
|
return fmt.Errorf("record agent session head: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// CreateTaskAttemptPrompt atomically creates all three records. The prompt is
|
|
// durable and the attempt head points to it before the transaction becomes
|
|
// visible, so an orchestrator may safely deliver the prompt after this returns.
|
|
func (s *Store) CreateTaskAttemptPrompt(
|
|
ctx context.Context,
|
|
task Task,
|
|
attempt Attempt,
|
|
prompt State,
|
|
parents []Parent,
|
|
) (Task, Attempt, State, error) {
|
|
now := time.Now().UTC()
|
|
if task.ID == "" {
|
|
task.ID = newID("t")
|
|
}
|
|
if task.CreatedAt.IsZero() {
|
|
task.CreatedAt = now
|
|
}
|
|
if task.Status == "" {
|
|
task.Status = "active"
|
|
}
|
|
if task.Title == "" {
|
|
task.Title = prompt.Prompt
|
|
}
|
|
task.Title, _ = RedactPromptSecrets(task.Title)
|
|
if attempt.ID == "" {
|
|
attempt.ID = newID("at")
|
|
}
|
|
if attempt.TaskID == "" {
|
|
attempt.TaskID = task.ID
|
|
}
|
|
if attempt.TaskID != task.ID {
|
|
return Task{}, Attempt{}, State{}, errors.New("hop: attempt task does not match task")
|
|
}
|
|
if attempt.CreatedAt.IsZero() {
|
|
attempt.CreatedAt = now
|
|
}
|
|
if attempt.Status == "" {
|
|
attempt.Status = "active"
|
|
}
|
|
if prompt.ID == "" {
|
|
prompt.ID = newID("p")
|
|
}
|
|
if prompt.Kind == "" {
|
|
prompt.Kind = StatePrompt
|
|
}
|
|
if prompt.Kind != StatePrompt {
|
|
return Task{}, Attempt{}, State{}, fmt.Errorf("hop: task instruction must have kind %q", StatePrompt)
|
|
}
|
|
if prompt.TaskID == "" {
|
|
prompt.TaskID = task.ID
|
|
}
|
|
if prompt.TaskID != task.ID {
|
|
return Task{}, Attempt{}, State{}, errors.New("hop: prompt task does not match task")
|
|
}
|
|
if prompt.AttemptID == "" {
|
|
prompt.AttemptID = attempt.ID
|
|
}
|
|
if prompt.AttemptID != attempt.ID {
|
|
return Task{}, Attempt{}, State{}, errors.New("hop: prompt attempt does not match attempt")
|
|
}
|
|
if prompt.Agent == "" {
|
|
prompt.Agent = attempt.Agent
|
|
}
|
|
if prompt.CreatedAt.IsZero() {
|
|
prompt.CreatedAt = now
|
|
}
|
|
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Task{}, Attempt{}, State{}, fmt.Errorf("begin task creation: %w", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
acceptedHead, err := metaTx(ctx, tx, "accepted_head")
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return Task{}, Attempt{}, State{}, ErrNotInitialized
|
|
}
|
|
if err != nil {
|
|
return Task{}, Attempt{}, State{}, fmt.Errorf("read accepted head: %w", err)
|
|
}
|
|
if attempt.BaseStateID == "" {
|
|
attempt.BaseStateID = acceptedHead
|
|
}
|
|
base, err := stateTx(ctx, tx, attempt.BaseStateID)
|
|
if err != nil {
|
|
return Task{}, Attempt{}, State{}, fmt.Errorf("read attempt base state: %w", err)
|
|
}
|
|
if attempt.HeadStateID != "" && attempt.HeadStateID != prompt.ID {
|
|
return Task{}, Attempt{}, State{}, errors.New("hop: new attempt head must be the prompt state")
|
|
}
|
|
attempt.HeadStateID = prompt.ID
|
|
if prompt.CanonicalAnchorID == "" {
|
|
prompt.CanonicalAnchorID = acceptedHead
|
|
}
|
|
if prompt.SourceTree == "" {
|
|
prompt.SourceTree = base.SourceTree
|
|
}
|
|
if prompt.GitCommit == "" {
|
|
prompt.GitCommit = base.GitCommit
|
|
}
|
|
|
|
parents = chooseParents(prompt.Parents, parents)
|
|
parents = ensureParentRole(parents, "run_parent", attempt.BaseStateID)
|
|
parents = ensureParentRole(parents, "canonical_anchor", prompt.CanonicalAnchorID)
|
|
parents = canonicalizeParents(parents)
|
|
prompt.Parents = parents
|
|
|
|
if _, err := tx.ExecContext(ctx,
|
|
`INSERT INTO tasks(id, title, status, created_at) VALUES (?, ?, ?, ?)`,
|
|
task.ID, task.Title, task.Status, formatTime(task.CreatedAt)); err != nil {
|
|
return Task{}, Attempt{}, State{}, fmt.Errorf("insert task: %w", err)
|
|
}
|
|
// head_state_id is temporarily NULL to break the intentional attempt/state
|
|
// reference cycle; it is populated before this transaction commits.
|
|
if _, err := tx.ExecContext(ctx,
|
|
`INSERT INTO attempts(id, task_id, agent, workspace, base_state_id, head_state_id, status, created_at)
|
|
VALUES (?, ?, ?, ?, ?, NULL, ?, ?)`,
|
|
attempt.ID, attempt.TaskID, attempt.Agent, attempt.Workspace, attempt.BaseStateID,
|
|
attempt.Status, formatTime(attempt.CreatedAt)); err != nil {
|
|
return Task{}, Attempt{}, State{}, fmt.Errorf("insert attempt: %w", err)
|
|
}
|
|
if err := insertStateTx(ctx, tx, prompt, parents); err != nil {
|
|
return Task{}, Attempt{}, State{}, err
|
|
}
|
|
if _, err := tx.ExecContext(ctx,
|
|
`UPDATE attempts SET head_state_id = ? WHERE id = ?`, prompt.ID, attempt.ID); err != nil {
|
|
return Task{}, Attempt{}, State{}, fmt.Errorf("install initial attempt head: %w", err)
|
|
}
|
|
if err := insertEventTx(ctx, tx, Event{
|
|
Kind: "task.created", TaskID: task.ID,
|
|
}, map[string]string{"title": task.Title}); err != nil {
|
|
return Task{}, Attempt{}, State{}, err
|
|
}
|
|
if err := insertEventTx(ctx, tx, Event{
|
|
Kind: "prompt.created", TaskID: task.ID, AttemptID: attempt.ID, StateID: prompt.ID,
|
|
}, nil); err != nil {
|
|
return Task{}, Attempt{}, State{}, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return Task{}, Attempt{}, State{}, fmt.Errorf("commit task creation: %w", err)
|
|
}
|
|
return task, attempt, prompt, nil
|
|
}
|
|
|
|
// AppendState atomically appends an immutable state and compare-and-swaps the
|
|
// owning attempt's head. An empty expectedAttemptHeadID means "the head observed
|
|
// by this transaction"; callers that already observed a head should pass it.
|
|
func (s *Store) AppendState(
|
|
ctx context.Context,
|
|
state State,
|
|
parents []Parent,
|
|
expectedAttemptHeadID string,
|
|
) (State, error) {
|
|
if state.AttemptID == "" {
|
|
return State{}, errors.New("hop: appended state requires an attempt ID")
|
|
}
|
|
if state.ID == "" {
|
|
state.ID = newID(stateIDPrefix(state.Kind))
|
|
}
|
|
if state.Kind == "" {
|
|
state.Kind = StateCheckpoint
|
|
}
|
|
if state.CreatedAt.IsZero() {
|
|
state.CreatedAt = time.Now().UTC()
|
|
}
|
|
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("begin state append: %w", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
attempt, err := attemptTx(ctx, tx, state.AttemptID)
|
|
if err != nil {
|
|
return State{}, err
|
|
}
|
|
if expectedAttemptHeadID == "" {
|
|
expectedAttemptHeadID = attempt.HeadStateID
|
|
}
|
|
if attempt.HeadStateID != expectedAttemptHeadID {
|
|
return State{}, &HeadChangedError{Scope: "attempt", Expected: expectedAttemptHeadID, Actual: attempt.HeadStateID}
|
|
}
|
|
if state.TaskID == "" {
|
|
state.TaskID = attempt.TaskID
|
|
}
|
|
if state.TaskID != attempt.TaskID {
|
|
return State{}, errors.New("hop: appended state task does not match attempt")
|
|
}
|
|
if state.Agent == "" {
|
|
state.Agent = attempt.Agent
|
|
}
|
|
head, err := stateTx(ctx, tx, attempt.HeadStateID)
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("read attempt head state: %w", err)
|
|
}
|
|
if state.SourceTree == "" {
|
|
state.SourceTree = head.SourceTree
|
|
}
|
|
if state.GitCommit == "" {
|
|
state.GitCommit = head.GitCommit
|
|
}
|
|
if state.CanonicalAnchorID == "" {
|
|
state.CanonicalAnchorID = head.CanonicalAnchorID
|
|
if state.CanonicalAnchorID == "" {
|
|
state.CanonicalAnchorID = attempt.BaseStateID
|
|
}
|
|
}
|
|
parents = chooseParents(state.Parents, parents)
|
|
parents = ensureParentRole(parents, "run_parent", expectedAttemptHeadID)
|
|
if state.Kind == StatePrompt {
|
|
parents = ensureParentRole(parents, "canonical_anchor", state.CanonicalAnchorID)
|
|
}
|
|
parents = canonicalizeParents(parents)
|
|
state.Parents = parents
|
|
|
|
if err := insertStateTx(ctx, tx, state, parents); err != nil {
|
|
return State{}, err
|
|
}
|
|
result, err := tx.ExecContext(ctx,
|
|
`UPDATE attempts SET head_state_id = ? WHERE id = ? AND head_state_id = ?`,
|
|
state.ID, state.AttemptID, expectedAttemptHeadID)
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("advance attempt head: %w", err)
|
|
}
|
|
changed, err := result.RowsAffected()
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("inspect attempt head update: %w", err)
|
|
}
|
|
if changed != 1 {
|
|
actual, _ := attemptHeadTx(ctx, tx, state.AttemptID)
|
|
return State{}, &HeadChangedError{Scope: "attempt", Expected: expectedAttemptHeadID, Actual: actual}
|
|
}
|
|
if err := insertEventTx(ctx, tx, Event{
|
|
Kind: "state.appended", TaskID: state.TaskID, AttemptID: state.AttemptID, StateID: state.ID,
|
|
}, map[string]string{"kind": string(state.Kind)}); err != nil {
|
|
return State{}, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return State{}, fmt.Errorf("commit state append: %w", err)
|
|
}
|
|
return state, nil
|
|
}
|
|
|
|
// CASAccept atomically inserts an accepted state and advances accepted_head only
|
|
// if it still equals expectedHeadID. Any inserted state is rolled back when the
|
|
// compare-and-swap fails.
|
|
func (s *Store) CASAccept(
|
|
ctx context.Context,
|
|
expectedHeadID string,
|
|
accepted State,
|
|
parents []Parent,
|
|
) (State, error) {
|
|
if expectedHeadID == "" {
|
|
return State{}, errors.New("hop: expected accepted head is required")
|
|
}
|
|
if accepted.ID == "" {
|
|
accepted.ID = newID("a")
|
|
}
|
|
if accepted.Kind == "" {
|
|
accepted.Kind = StateAccepted
|
|
}
|
|
if accepted.Kind != StateAccepted {
|
|
return State{}, fmt.Errorf("hop: accepted state must have kind %q", StateAccepted)
|
|
}
|
|
if accepted.CreatedAt.IsZero() {
|
|
accepted.CreatedAt = time.Now().UTC()
|
|
}
|
|
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("begin acceptance: %w", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
actual, err := metaTx(ctx, tx, "accepted_head")
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return State{}, ErrNotInitialized
|
|
}
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("read accepted head: %w", err)
|
|
}
|
|
if actual != expectedHeadID {
|
|
return State{}, &HeadChangedError{Scope: "accepted", Expected: expectedHeadID, Actual: actual}
|
|
}
|
|
current, err := stateTx(ctx, tx, expectedHeadID)
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("read current accepted state: %w", err)
|
|
}
|
|
|
|
parents = chooseParents(accepted.Parents, parents)
|
|
parents = ensureParentRole(parents, "canonical_parent", expectedHeadID)
|
|
parents = canonicalizeParents(parents)
|
|
accepted.Parents = parents
|
|
if accepted.CanonicalAnchorID == "" {
|
|
accepted.CanonicalAnchorID = expectedHeadID
|
|
}
|
|
|
|
// When the caller supplied a proposal parent but omitted repeated fields,
|
|
// inherit them from that immutable proposal.
|
|
if proposalParent, ok := parentWithRole(parents, "proposal_parent"); ok {
|
|
proposal, err := stateTx(ctx, tx, proposalParent.StateID)
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("read proposal parent: %w", err)
|
|
}
|
|
if accepted.TaskID == "" {
|
|
accepted.TaskID = proposal.TaskID
|
|
}
|
|
if accepted.AttemptID == "" {
|
|
accepted.AttemptID = proposal.AttemptID
|
|
}
|
|
if accepted.SourceTree == "" {
|
|
accepted.SourceTree = proposal.SourceTree
|
|
}
|
|
if accepted.GitCommit == "" {
|
|
accepted.GitCommit = proposal.GitCommit
|
|
}
|
|
if accepted.Agent == "" {
|
|
accepted.Agent = proposal.Agent
|
|
}
|
|
}
|
|
if accepted.SourceTree == "" {
|
|
accepted.SourceTree = current.SourceTree
|
|
}
|
|
if accepted.GitCommit == "" {
|
|
accepted.GitCommit = current.GitCommit
|
|
}
|
|
|
|
if err := insertStateTx(ctx, tx, accepted, parents); err != nil {
|
|
return State{}, err
|
|
}
|
|
result, err := tx.ExecContext(ctx,
|
|
`UPDATE meta SET value = ? WHERE key = 'accepted_head' AND value = ?`,
|
|
accepted.ID, expectedHeadID)
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("advance accepted head: %w", err)
|
|
}
|
|
changed, err := result.RowsAffected()
|
|
if err != nil {
|
|
return State{}, fmt.Errorf("inspect accepted head update: %w", err)
|
|
}
|
|
if changed != 1 {
|
|
actual, _ := metaTx(ctx, tx, "accepted_head")
|
|
return State{}, &HeadChangedError{Scope: "accepted", Expected: expectedHeadID, Actual: actual}
|
|
}
|
|
if accepted.AttemptID != "" {
|
|
if _, err := tx.ExecContext(ctx,
|
|
`UPDATE attempts SET status = 'accepted' WHERE id = ?`, accepted.AttemptID); err != nil {
|
|
return State{}, fmt.Errorf("mark attempt accepted: %w", err)
|
|
}
|
|
}
|
|
if accepted.TaskID != "" {
|
|
if _, err := tx.ExecContext(ctx,
|
|
`UPDATE tasks SET status = 'accepted' WHERE id = ?`, accepted.TaskID); err != nil {
|
|
return State{}, fmt.Errorf("mark task accepted: %w", err)
|
|
}
|
|
}
|
|
if err := insertEventTx(ctx, tx, Event{
|
|
Kind: "state.accepted", TaskID: accepted.TaskID, AttemptID: accepted.AttemptID, StateID: accepted.ID,
|
|
}, map[string]string{"previous_head": expectedHeadID}); err != nil {
|
|
return State{}, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return State{}, fmt.Errorf("commit acceptance: %w", err)
|
|
}
|
|
return accepted, nil
|
|
}
|
|
|
|
// AcceptedHeadID returns the current canonical pointer without loading the
|
|
// state body.
|
|
func (s *Store) AcceptedHeadID(ctx context.Context) (string, error) {
|
|
var id string
|
|
err := s.db.QueryRowContext(ctx, `SELECT value FROM meta WHERE key = 'accepted_head'`).Scan(&id)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", ErrNotInitialized
|
|
}
|
|
if err != nil {
|
|
return "", fmt.Errorf("read accepted head: %w", err)
|
|
}
|
|
return id, nil
|
|
}
|
|
|
|
func (s *Store) AcceptedHead(ctx context.Context) (State, error) {
|
|
id, err := s.AcceptedHeadID(ctx)
|
|
if err != nil {
|
|
return State{}, err
|
|
}
|
|
return s.GetState(ctx, id)
|
|
}
|
|
|
|
func (s *Store) RepositoryRoot(ctx context.Context) (string, error) {
|
|
var root string
|
|
err := s.db.QueryRowContext(ctx, `SELECT value FROM meta WHERE key = 'root'`).Scan(&root)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", ErrNotInitialized
|
|
}
|
|
if err != nil {
|
|
return "", fmt.Errorf("read repository root: %w", err)
|
|
}
|
|
return root, nil
|
|
}
|
|
|
|
func (s *Store) GetState(ctx context.Context, id string) (State, error) {
|
|
state, err := scanState(s.db.QueryRowContext(ctx, `SELECT `+stateColumns+` FROM states WHERE id = ?`, id))
|
|
if err != nil {
|
|
return State{}, dbNotFound("state", id, err)
|
|
}
|
|
parents, err := s.GetParents(ctx, id)
|
|
if err != nil {
|
|
return State{}, err
|
|
}
|
|
state.Parents = parents
|
|
return state, nil
|
|
}
|
|
|
|
func (s *Store) GetParents(ctx context.Context, stateID string) ([]Parent, error) {
|
|
rows, err := s.db.QueryContext(ctx,
|
|
`SELECT parent_state_id, role, parent_order
|
|
FROM state_parents WHERE state_id = ? ORDER BY parent_order`, stateID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query state parents: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var parents []Parent
|
|
for rows.Next() {
|
|
var parent Parent
|
|
if err := rows.Scan(&parent.StateID, &parent.Role, &parent.Order); err != nil {
|
|
return nil, fmt.Errorf("scan state parent: %w", err)
|
|
}
|
|
parents = append(parents, parent)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate state parents: %w", err)
|
|
}
|
|
return parents, nil
|
|
}
|
|
|
|
func (s *Store) ParentByRole(ctx context.Context, stateID, role string) (Parent, error) {
|
|
var parent Parent
|
|
err := s.db.QueryRowContext(ctx,
|
|
`SELECT parent_state_id, role, parent_order
|
|
FROM state_parents WHERE state_id = ? AND role = ? ORDER BY parent_order LIMIT 1`,
|
|
stateID, role).Scan(&parent.StateID, &parent.Role, &parent.Order)
|
|
if err != nil {
|
|
return Parent{}, dbNotFound("parent", role, err)
|
|
}
|
|
return parent, nil
|
|
}
|
|
|
|
func (s *Store) ParentsByRole(ctx context.Context, stateID, role string) ([]Parent, error) {
|
|
rows, err := s.db.QueryContext(ctx,
|
|
`SELECT parent_state_id, role, parent_order
|
|
FROM state_parents WHERE state_id = ? AND role = ? ORDER BY parent_order`,
|
|
stateID, role)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query state parents by role: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var parents []Parent
|
|
for rows.Next() {
|
|
var parent Parent
|
|
if err := rows.Scan(&parent.StateID, &parent.Role, &parent.Order); err != nil {
|
|
return nil, fmt.Errorf("scan state parent: %w", err)
|
|
}
|
|
parents = append(parents, parent)
|
|
}
|
|
return parents, rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetTask(ctx context.Context, id string) (Task, error) {
|
|
return scanTask(s.db.QueryRowContext(ctx,
|
|
`SELECT id, title, status, created_at FROM tasks WHERE id = ?`, id), id)
|
|
}
|
|
|
|
func (s *Store) ListTasks(ctx context.Context, status string) ([]Task, error) {
|
|
query := `SELECT id, title, status, created_at FROM tasks`
|
|
var args []any
|
|
if status != "" {
|
|
query += ` WHERE status = ?`
|
|
args = append(args, status)
|
|
}
|
|
query += ` ORDER BY created_at, id`
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query tasks: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var tasks []Task
|
|
for rows.Next() {
|
|
task, err := scanTask(rows, "")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tasks = append(tasks, task)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate tasks: %w", err)
|
|
}
|
|
return tasks, nil
|
|
}
|
|
|
|
func (s *Store) GetAttempt(ctx context.Context, id string) (Attempt, error) {
|
|
attempt, err := scanAttempt(s.db.QueryRowContext(ctx,
|
|
`SELECT `+attemptColumns+` FROM attempts WHERE id = ?`, id))
|
|
if err != nil {
|
|
return Attempt{}, dbNotFound("attempt", id, err)
|
|
}
|
|
return attempt, nil
|
|
}
|
|
|
|
// ListAttempts filters by task and status when either value is non-empty.
|
|
func (s *Store) ListAttempts(ctx context.Context, taskID, status string) ([]Attempt, error) {
|
|
query := `SELECT ` + attemptColumns + ` FROM attempts WHERE 1 = 1`
|
|
var args []any
|
|
if taskID != "" {
|
|
query += ` AND task_id = ?`
|
|
args = append(args, taskID)
|
|
}
|
|
if status != "" {
|
|
query += ` AND status = ?`
|
|
args = append(args, status)
|
|
}
|
|
query += ` ORDER BY created_at, id`
|
|
return s.queryAttempts(ctx, query, args...)
|
|
}
|
|
|
|
func (s *Store) ActiveAttempts(ctx context.Context) ([]Attempt, error) {
|
|
return s.queryAttempts(ctx,
|
|
`SELECT `+attemptColumns+` FROM attempts
|
|
WHERE status NOT IN ('accepted', 'completed', 'failed', 'cancelled', 'rejected')
|
|
ORDER BY created_at, id`)
|
|
}
|
|
|
|
func (s *Store) queryAttempts(ctx context.Context, query string, args ...any) ([]Attempt, error) {
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query attempts: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var attempts []Attempt
|
|
for rows.Next() {
|
|
attempt, err := scanAttempt(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
attempts = append(attempts, attempt)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate attempts: %w", err)
|
|
}
|
|
return attempts, nil
|
|
}
|
|
|
|
func (s *Store) UpdateAttemptStatus(ctx context.Context, id, status string) error {
|
|
if status == "" {
|
|
return errors.New("hop: attempt status is required")
|
|
}
|
|
result, err := s.db.ExecContext(ctx, `UPDATE attempts SET status = ? WHERE id = ?`, status, id)
|
|
if err != nil {
|
|
return fmt.Errorf("update attempt status: %w", err)
|
|
}
|
|
return requireOneRow(result, "attempt", id)
|
|
}
|
|
|
|
func (s *Store) UpdateTaskStatus(ctx context.Context, id, status string) error {
|
|
if status == "" {
|
|
return errors.New("hop: task status is required")
|
|
}
|
|
result, err := s.db.ExecContext(ctx, `UPDATE tasks SET status = ? WHERE id = ?`, status, id)
|
|
if err != nil {
|
|
return fmt.Errorf("update task status: %w", err)
|
|
}
|
|
return requireOneRow(result, "task", id)
|
|
}
|
|
|
|
// UpdateAttemptHead performs a standalone compare-and-swap. AppendState should
|
|
// normally be preferred because it cannot point at a state that was not created
|
|
// as part of the same logical operation.
|
|
func (s *Store) UpdateAttemptHead(ctx context.Context, attemptID, expectedHeadID, nextHeadID string) error {
|
|
result, err := s.db.ExecContext(ctx,
|
|
`UPDATE attempts SET head_state_id = ? WHERE id = ? AND head_state_id = ?`,
|
|
nextHeadID, attemptID, expectedHeadID)
|
|
if err != nil {
|
|
return fmt.Errorf("advance attempt head: %w", err)
|
|
}
|
|
changed, err := result.RowsAffected()
|
|
if err != nil {
|
|
return fmt.Errorf("inspect attempt head update: %w", err)
|
|
}
|
|
if changed == 1 {
|
|
return nil
|
|
}
|
|
attempt, getErr := s.GetAttempt(ctx, attemptID)
|
|
if getErr != nil {
|
|
return getErr
|
|
}
|
|
return &HeadChangedError{Scope: "attempt", Expected: expectedHeadID, Actual: attempt.HeadStateID}
|
|
}
|
|
|
|
// AddCheck records evidence bound to a tree. If StateID is present, TreeHash is
|
|
// filled from that state when empty and otherwise must match its source tree.
|
|
func (s *Store) AddCheck(ctx context.Context, check Check) (Check, error) {
|
|
if check.ID == "" {
|
|
check.ID = newID("check")
|
|
}
|
|
if check.AttemptID == "" {
|
|
return Check{}, errors.New("hop: check attempt ID is required")
|
|
}
|
|
if check.CreatedAt.IsZero() {
|
|
check.CreatedAt = time.Now().UTC()
|
|
}
|
|
check.Command, _ = redactSecretStrings(check.Command)
|
|
check.Output, _ = RedactPromptSecrets(check.Output)
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Check{}, fmt.Errorf("begin check creation: %w", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
if _, err := attemptTx(ctx, tx, check.AttemptID); err != nil {
|
|
return Check{}, err
|
|
}
|
|
if check.StateID != "" {
|
|
state, err := stateTx(ctx, tx, check.StateID)
|
|
if err != nil {
|
|
return Check{}, err
|
|
}
|
|
if state.AttemptID != "" && state.AttemptID != check.AttemptID {
|
|
return Check{}, errors.New("hop: check state belongs to another attempt")
|
|
}
|
|
if check.TreeHash == "" {
|
|
check.TreeHash = state.SourceTree
|
|
} else if check.TreeHash != state.SourceTree {
|
|
return Check{}, errors.New("hop: check tree hash does not match state source tree")
|
|
}
|
|
}
|
|
if check.TreeHash == "" {
|
|
return Check{}, errors.New("hop: check tree hash is required")
|
|
}
|
|
command, err := json.Marshal(check.Command)
|
|
if err != nil {
|
|
return Check{}, fmt.Errorf("encode check command: %w", err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx,
|
|
`INSERT INTO checks(id, attempt_id, state_id, tree_hash, command_json, exit_code, output, created_at)
|
|
VALUES (?, ?, NULLIF(?, ''), ?, ?, ?, ?, ?)`,
|
|
check.ID, check.AttemptID, check.StateID, check.TreeHash, string(command), check.ExitCode,
|
|
check.Output, formatTime(check.CreatedAt)); err != nil {
|
|
return Check{}, fmt.Errorf("insert check: %w", err)
|
|
}
|
|
attempt, err := attemptTx(ctx, tx, check.AttemptID)
|
|
if err != nil {
|
|
return Check{}, err
|
|
}
|
|
if err := insertEventTx(ctx, tx, Event{
|
|
Kind: "check.recorded", TaskID: attempt.TaskID, AttemptID: check.AttemptID, StateID: check.StateID,
|
|
}, map[string]any{"check_id": check.ID, "tree_hash": check.TreeHash, "exit_code": check.ExitCode}); err != nil {
|
|
return Check{}, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return Check{}, fmt.Errorf("commit check creation: %w", err)
|
|
}
|
|
return check, nil
|
|
}
|
|
|
|
// CreateCheck is a convenience alias for AddCheck.
|
|
func (s *Store) CreateCheck(ctx context.Context, check Check) (Check, error) {
|
|
return s.AddCheck(ctx, check)
|
|
}
|
|
|
|
func (s *Store) GetCheck(ctx context.Context, id string) (Check, error) {
|
|
check, err := scanCheck(s.db.QueryRowContext(ctx,
|
|
`SELECT `+checkColumns+` FROM checks WHERE id = ?`, id))
|
|
if err != nil {
|
|
return Check{}, dbNotFound("check", id, err)
|
|
}
|
|
return check, nil
|
|
}
|
|
|
|
// ListChecks filters by attempt and/or exact source tree. Passing both empty
|
|
// lists all checks.
|
|
func (s *Store) ListChecks(ctx context.Context, attemptID, treeHash string) ([]Check, error) {
|
|
query := `SELECT ` + checkColumns + ` FROM checks WHERE 1 = 1`
|
|
var args []any
|
|
if attemptID != "" {
|
|
query += ` AND attempt_id = ?`
|
|
args = append(args, attemptID)
|
|
}
|
|
if treeHash != "" {
|
|
query += ` AND tree_hash = ?`
|
|
args = append(args, treeHash)
|
|
}
|
|
query += ` ORDER BY created_at, id`
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query checks: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var checks []Check
|
|
for rows.Next() {
|
|
check, err := scanCheck(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
checks = append(checks, check)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate checks: %w", err)
|
|
}
|
|
return checks, nil
|
|
}
|
|
|
|
func (s *Store) ChecksForTree(ctx context.Context, treeHash string) ([]Check, error) {
|
|
return s.ListChecks(ctx, "", treeHash)
|
|
}
|
|
|
|
// Graph returns states in creation order with their typed parent edges. An
|
|
// empty taskID returns the repository-wide graph.
|
|
func (s *Store) Graph(ctx context.Context, taskID string) ([]GraphRow, error) {
|
|
query := `SELECT ` + stateColumns + ` FROM states`
|
|
var args []any
|
|
if taskID != "" {
|
|
query += ` WHERE task_id = ?`
|
|
args = append(args, taskID)
|
|
}
|
|
query += ` ORDER BY created_at, id`
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query state graph: %w", err)
|
|
}
|
|
var states []State
|
|
for rows.Next() {
|
|
state, err := scanState(rows)
|
|
if err != nil {
|
|
_ = rows.Close()
|
|
return nil, err
|
|
}
|
|
states = append(states, state)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
_ = rows.Close()
|
|
return nil, fmt.Errorf("iterate state graph: %w", err)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, fmt.Errorf("close state graph rows: %w", err)
|
|
}
|
|
|
|
graph := make([]GraphRow, 0, len(states))
|
|
for _, state := range states {
|
|
parents, err := s.GetParents(ctx, state.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
state.Parents = parents
|
|
graph = append(graph, GraphRow{State: state, Parents: parents})
|
|
}
|
|
return graph, nil
|
|
}
|
|
|
|
// AcceptedHistory follows canonical_parent edges from newest to oldest.
|
|
func (s *Store) AcceptedHistory(ctx context.Context, limit int) ([]State, error) {
|
|
if limit <= 0 {
|
|
limit = 1000
|
|
}
|
|
rows, err := s.db.QueryContext(ctx,
|
|
`WITH RECURSIVE lineage(id, depth) AS (
|
|
SELECT value, 0 FROM meta WHERE key = 'accepted_head'
|
|
UNION ALL
|
|
SELECT p.parent_state_id, lineage.depth + 1
|
|
FROM lineage
|
|
JOIN state_parents p ON p.state_id = lineage.id AND p.role = 'canonical_parent'
|
|
WHERE lineage.depth < 9999
|
|
)
|
|
SELECT id FROM lineage ORDER BY depth LIMIT ?`, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query accepted history: %w", err)
|
|
}
|
|
var ids []string
|
|
for rows.Next() {
|
|
var id string
|
|
if err := rows.Scan(&id); err != nil {
|
|
_ = rows.Close()
|
|
return nil, fmt.Errorf("scan accepted history: %w", err)
|
|
}
|
|
ids = append(ids, id)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
_ = rows.Close()
|
|
return nil, fmt.Errorf("iterate accepted history: %w", err)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, fmt.Errorf("close accepted history rows: %w", err)
|
|
}
|
|
if len(ids) == 0 {
|
|
return nil, ErrNotInitialized
|
|
}
|
|
history := make([]State, 0, len(ids))
|
|
for _, id := range ids {
|
|
state, err := s.GetState(ctx, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
history = append(history, state)
|
|
}
|
|
return history, nil
|
|
}
|
|
|
|
// History is a convenience alias for AcceptedHistory.
|
|
func (s *Store) History(ctx context.Context, limit int) ([]State, error) {
|
|
return s.AcceptedHistory(ctx, limit)
|
|
}
|
|
|
|
func (s *Store) Status(ctx context.Context) (Status, error) {
|
|
root, err := s.RepositoryRoot(ctx)
|
|
if err != nil {
|
|
return Status{}, err
|
|
}
|
|
head, err := s.AcceptedHead(ctx)
|
|
if err != nil {
|
|
return Status{}, err
|
|
}
|
|
attempts, err := s.ActiveAttempts(ctx)
|
|
if err != nil {
|
|
return Status{}, err
|
|
}
|
|
return Status{Root: root, AcceptedHead: head, Attempts: attempts}, nil
|
|
}
|
|
|
|
// AppendEvent adds an explicit audit event. Store transitions also add their
|
|
// own events in the same transactions as the state changes they describe.
|
|
func (s *Store) AppendEvent(ctx context.Context, event Event) (Event, error) {
|
|
if event.Kind == "" {
|
|
return Event{}, errors.New("hop: event kind is required")
|
|
}
|
|
if event.ID == "" {
|
|
event.ID = newID("e")
|
|
}
|
|
if event.CreatedAt.IsZero() {
|
|
event.CreatedAt = time.Now().UTC()
|
|
}
|
|
payload := event.Payload
|
|
if len(payload) == 0 {
|
|
payload = json.RawMessage(`{}`)
|
|
}
|
|
if !json.Valid(payload) {
|
|
return Event{}, errors.New("hop: event payload is not valid JSON")
|
|
}
|
|
_, err := s.db.ExecContext(ctx,
|
|
`INSERT INTO events(id, kind, task_id, attempt_id, state_id, payload_json, created_at)
|
|
VALUES (?, ?, NULLIF(?, ''), NULLIF(?, ''), NULLIF(?, ''), ?, ?)`,
|
|
event.ID, event.Kind, event.TaskID, event.AttemptID, event.StateID, string(payload), formatTime(event.CreatedAt))
|
|
if err != nil {
|
|
return Event{}, fmt.Errorf("insert event: %w", err)
|
|
}
|
|
event.Payload = payload
|
|
return event, nil
|
|
}
|
|
|
|
func (s *Store) ListEvents(ctx context.Context, limit int) ([]Event, error) {
|
|
if limit <= 0 {
|
|
limit = 1000
|
|
}
|
|
rows, err := s.db.QueryContext(ctx,
|
|
`SELECT id, kind, COALESCE(task_id, ''), COALESCE(attempt_id, ''), COALESCE(state_id, ''), payload_json, created_at
|
|
FROM events ORDER BY created_at, id LIMIT ?`, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("query events: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var events []Event
|
|
for rows.Next() {
|
|
var event Event
|
|
var payload, created string
|
|
if err := rows.Scan(&event.ID, &event.Kind, &event.TaskID, &event.AttemptID, &event.StateID, &payload, &created); err != nil {
|
|
return nil, fmt.Errorf("scan event: %w", err)
|
|
}
|
|
createdAt, err := parseTime(created)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
event.CreatedAt = createdAt
|
|
event.Payload = json.RawMessage(payload)
|
|
events = append(events, event)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate events: %w", err)
|
|
}
|
|
return events, nil
|
|
}
|
|
|
|
const stateColumns = `id, kind, COALESCE(task_id, ''), COALESCE(attempt_id, ''),
|
|
COALESCE(canonical_anchor_id, ''), source_tree, git_commit, prompt, summary, agent, digest, created_at`
|
|
|
|
const attemptColumns = `id, task_id, agent, workspace, base_state_id, COALESCE(head_state_id, ''), status, created_at`
|
|
|
|
const checkColumns = `id, attempt_id, COALESCE(state_id, ''), tree_hash, command_json, exit_code, output, created_at`
|
|
|
|
type scanner interface {
|
|
Scan(dest ...any) error
|
|
}
|
|
|
|
func scanState(row scanner) (State, error) {
|
|
var state State
|
|
var kind, created string
|
|
if err := row.Scan(
|
|
&state.ID, &kind, &state.TaskID, &state.AttemptID, &state.CanonicalAnchorID,
|
|
&state.SourceTree, &state.GitCommit, &state.Prompt, &state.Summary, &state.Agent,
|
|
&state.Digest, &created,
|
|
); err != nil {
|
|
return State{}, err
|
|
}
|
|
createdAt, err := parseTime(created)
|
|
if err != nil {
|
|
return State{}, err
|
|
}
|
|
state.Kind = StateKind(kind)
|
|
state.CreatedAt = createdAt
|
|
return state, nil
|
|
}
|
|
|
|
func scanTask(row scanner, requestedID string) (Task, error) {
|
|
var task Task
|
|
var created string
|
|
if err := row.Scan(&task.ID, &task.Title, &task.Status, &created); err != nil {
|
|
return Task{}, dbNotFound("task", requestedID, err)
|
|
}
|
|
createdAt, err := parseTime(created)
|
|
if err != nil {
|
|
return Task{}, err
|
|
}
|
|
task.CreatedAt = createdAt
|
|
return task, nil
|
|
}
|
|
|
|
func scanAttempt(row scanner) (Attempt, error) {
|
|
var attempt Attempt
|
|
var created string
|
|
if err := row.Scan(
|
|
&attempt.ID, &attempt.TaskID, &attempt.Agent, &attempt.Workspace,
|
|
&attempt.BaseStateID, &attempt.HeadStateID, &attempt.Status, &created,
|
|
); err != nil {
|
|
return Attempt{}, err
|
|
}
|
|
createdAt, err := parseTime(created)
|
|
if err != nil {
|
|
return Attempt{}, err
|
|
}
|
|
attempt.CreatedAt = createdAt
|
|
return attempt, nil
|
|
}
|
|
|
|
func scanCheck(row scanner) (Check, error) {
|
|
var check Check
|
|
var command, created string
|
|
if err := row.Scan(
|
|
&check.ID, &check.AttemptID, &check.StateID, &check.TreeHash, &command,
|
|
&check.ExitCode, &check.Output, &created,
|
|
); err != nil {
|
|
return Check{}, err
|
|
}
|
|
if err := json.Unmarshal([]byte(command), &check.Command); err != nil {
|
|
return Check{}, fmt.Errorf("decode check command: %w", err)
|
|
}
|
|
createdAt, err := parseTime(created)
|
|
if err != nil {
|
|
return Check{}, err
|
|
}
|
|
check.CreatedAt = createdAt
|
|
return check, nil
|
|
}
|
|
|
|
func stateTx(ctx context.Context, tx *sql.Tx, id string) (State, error) {
|
|
state, err := scanState(tx.QueryRowContext(ctx,
|
|
`SELECT `+stateColumns+` FROM states WHERE id = ?`, id))
|
|
if err != nil {
|
|
return State{}, dbNotFound("state", id, err)
|
|
}
|
|
return state, nil
|
|
}
|
|
|
|
func attemptTx(ctx context.Context, tx *sql.Tx, id string) (Attempt, error) {
|
|
attempt, err := scanAttempt(tx.QueryRowContext(ctx,
|
|
`SELECT `+attemptColumns+` FROM attempts WHERE id = ?`, id))
|
|
if err != nil {
|
|
return Attempt{}, dbNotFound("attempt", id, err)
|
|
}
|
|
return attempt, nil
|
|
}
|
|
|
|
func attemptHeadTx(ctx context.Context, tx *sql.Tx, id string) (string, error) {
|
|
var head string
|
|
err := tx.QueryRowContext(ctx,
|
|
`SELECT COALESCE(head_state_id, '') FROM attempts WHERE id = ?`, id).Scan(&head)
|
|
return head, err
|
|
}
|
|
|
|
func metaTx(ctx context.Context, tx *sql.Tx, key string) (string, error) {
|
|
var value string
|
|
err := tx.QueryRowContext(ctx, `SELECT value FROM meta WHERE key = ?`, key).Scan(&value)
|
|
return value, err
|
|
}
|
|
|
|
func insertStateTx(ctx context.Context, tx *sql.Tx, state State, parents []Parent) error {
|
|
if state.ID == "" {
|
|
return errors.New("hop: state ID is required")
|
|
}
|
|
if state.Kind == "" {
|
|
return errors.New("hop: state kind is required")
|
|
}
|
|
if state.CreatedAt.IsZero() {
|
|
return errors.New("hop: state creation time is required")
|
|
}
|
|
for name, value := range map[string]string{
|
|
"prompt": state.Prompt,
|
|
"summary": state.Summary,
|
|
"agent": state.Agent,
|
|
} {
|
|
if redacted, findings := RedactPromptSecrets(value); len(findings) > 0 || redacted != value {
|
|
return fmt.Errorf("hop: refusing to persist an unredacted credential in state %s", name)
|
|
}
|
|
}
|
|
_, err := tx.ExecContext(ctx,
|
|
`INSERT INTO states(
|
|
id, kind, task_id, attempt_id, canonical_anchor_id, source_tree, git_commit,
|
|
prompt, summary, agent, digest, created_at
|
|
) VALUES (?, ?, NULLIF(?, ''), NULLIF(?, ''), NULLIF(?, ''), ?, ?, ?, ?, ?, ?, ?)`,
|
|
state.ID, string(state.Kind), state.TaskID, state.AttemptID, state.CanonicalAnchorID,
|
|
state.SourceTree, state.GitCommit, state.Prompt, state.Summary, state.Agent,
|
|
state.Digest, formatTime(state.CreatedAt))
|
|
if err != nil {
|
|
return fmt.Errorf("insert state %q: %w", state.ID, err)
|
|
}
|
|
for _, parent := range parents {
|
|
if parent.StateID == "" {
|
|
return errors.New("hop: parent state ID is required")
|
|
}
|
|
if _, err := tx.ExecContext(ctx,
|
|
`INSERT INTO state_parents(state_id, parent_state_id, role, parent_order)
|
|
VALUES (?, ?, ?, ?)`, state.ID, parent.StateID, parent.Role, parent.Order); err != nil {
|
|
return fmt.Errorf("insert parent for state %q: %w", state.ID, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func insertEventTx(ctx context.Context, tx *sql.Tx, event Event, payload any) error {
|
|
if event.ID == "" {
|
|
event.ID = newID("e")
|
|
}
|
|
if event.CreatedAt.IsZero() {
|
|
event.CreatedAt = time.Now().UTC()
|
|
}
|
|
var encoded []byte
|
|
var err error
|
|
if payload == nil {
|
|
encoded = []byte(`{}`)
|
|
} else {
|
|
encoded, err = json.Marshal(payload)
|
|
if err != nil {
|
|
return fmt.Errorf("encode event payload: %w", err)
|
|
}
|
|
}
|
|
if redacted, findings := RedactPromptSecrets(string(encoded)); len(findings) > 0 || redacted != string(encoded) {
|
|
return fmt.Errorf("hop: refusing to persist an unredacted credential in event %q", event.Kind)
|
|
}
|
|
_, err = tx.ExecContext(ctx,
|
|
`INSERT INTO events(id, kind, task_id, attempt_id, state_id, payload_json, created_at)
|
|
VALUES (?, ?, NULLIF(?, ''), NULLIF(?, ''), NULLIF(?, ''), ?, ?)`,
|
|
event.ID, event.Kind, event.TaskID, event.AttemptID, event.StateID,
|
|
string(encoded), formatTime(event.CreatedAt))
|
|
if err != nil {
|
|
return fmt.Errorf("insert event %q: %w", event.Kind, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func chooseParents(fromState, explicit []Parent) []Parent {
|
|
if len(explicit) > 0 {
|
|
return append([]Parent(nil), explicit...)
|
|
}
|
|
return append([]Parent(nil), fromState...)
|
|
}
|
|
|
|
func ensureParentRole(parents []Parent, role, stateID string) []Parent {
|
|
if stateID == "" {
|
|
return parents
|
|
}
|
|
for _, parent := range parents {
|
|
if parent.Role == role {
|
|
return parents
|
|
}
|
|
}
|
|
return append(parents, Parent{StateID: stateID, Role: role, Order: len(parents)})
|
|
}
|
|
|
|
func parentWithRole(parents []Parent, role string) (Parent, bool) {
|
|
for _, parent := range parents {
|
|
if parent.Role == role {
|
|
return parent, true
|
|
}
|
|
}
|
|
return Parent{}, false
|
|
}
|
|
|
|
func canonicalizeParents(parents []Parent) []Parent {
|
|
parents = append([]Parent(nil), parents...)
|
|
used := make(map[int]bool, len(parents))
|
|
for i := range parents {
|
|
if parents[i].Role == "" {
|
|
parents[i].Role = "parent"
|
|
}
|
|
order := parents[i].Order
|
|
if order < 0 || used[order] {
|
|
order = 0
|
|
for used[order] {
|
|
order++
|
|
}
|
|
parents[i].Order = order
|
|
}
|
|
used[order] = true
|
|
}
|
|
return parents
|
|
}
|
|
|
|
func stateIDPrefix(kind StateKind) string {
|
|
switch kind {
|
|
case StatePrompt:
|
|
return "p"
|
|
case StateCheckpoint:
|
|
return "c"
|
|
case StateProposal:
|
|
return "r"
|
|
case StateAccepted:
|
|
return "a"
|
|
case StateFailed:
|
|
return "f"
|
|
case StateCancelled:
|
|
return "x"
|
|
default:
|
|
return "s"
|
|
}
|
|
}
|
|
|
|
func formatTime(value time.Time) string {
|
|
return value.UTC().Format(time.RFC3339Nano)
|
|
}
|
|
|
|
func parseTime(value string) (time.Time, error) {
|
|
parsed, err := time.Parse(time.RFC3339Nano, value)
|
|
if err != nil {
|
|
return time.Time{}, fmt.Errorf("decode database timestamp %q: %w", value, err)
|
|
}
|
|
return parsed, nil
|
|
}
|
|
|
|
func dbNotFound(kind, id string, err error) error {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
if id == "" {
|
|
return fmt.Errorf("%w: %s", ErrNotFound, kind)
|
|
}
|
|
return fmt.Errorf("%w: %s %q", ErrNotFound, kind, id)
|
|
}
|
|
return fmt.Errorf("read %s %q: %w", kind, id, err)
|
|
}
|
|
|
|
func requireOneRow(result sql.Result, kind, id string) error {
|
|
changed, err := result.RowsAffected()
|
|
if err != nil {
|
|
return fmt.Errorf("inspect %s update: %w", kind, err)
|
|
}
|
|
if changed != 1 {
|
|
return fmt.Errorf("%w: %s %q", ErrNotFound, kind, id)
|
|
}
|
|
return nil
|
|
}
|