547 lines
18 KiB
Go
547 lines
18 KiB
Go
package store
|
||
|
||
import (
|
||
"crypto/rand"
|
||
"database/sql"
|
||
"encoding/hex"
|
||
"errors"
|
||
"fmt"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
core "agent-system-core"
|
||
)
|
||
|
||
// isUniqueViolation 判定唯一约束冲突（PG 23505 / SQLite UNIQUE）。
|
||
func isUniqueViolation(err error) bool {
|
||
if err == nil {
|
||
return false
|
||
}
|
||
msg := err.Error()
|
||
return strings.Contains(msg, "23505") ||
|
||
strings.Contains(msg, "UNIQUE constraint") ||
|
||
strings.Contains(msg, "duplicate key value")
|
||
}
|
||
|
||
// ---- 消息 ----
|
||
|
||
// AppendMessage 以单语句原子分配会话内序号（INSERT ... SELECT MAX+1）；
|
||
// 并发同会话追加由 uq_messages_conv_seq 唯一约束兜底，冲突有界重试。
|
||
func (s *Store) AppendMessage(m *core.Message) error {
|
||
return s.AppendMessageWithRequest(m, "")
|
||
}
|
||
|
||
// AppendMessageWithRequest 同 AppendMessage，并记录可选的 request_id。
|
||
// 同一会话里非空 request_id 已存在时，把 *m 换成已有行并返回 DuplicateRequestError。
|
||
func (s *Store) AppendMessageWithRequest(m *core.Message, requestID string) error {
|
||
return s.appendMessageWithRequest(m, requestID, "", false, false)
|
||
}
|
||
|
||
// AppendMessageWithRequestAccountChecked fences the final user-message write
|
||
// against account disable. The actor lock precedes the conversation lock.
|
||
func (s *Store) AppendMessageWithRequestAccountChecked(actor string, allowImplicitDefault bool, m *core.Message, requestID string) error {
|
||
return s.appendMessageWithRequest(m, requestID, actor, allowImplicitDefault, true)
|
||
}
|
||
|
||
func (s *Store) appendMessageWithRequest(m *core.Message, requestID, actor string, allowImplicitDefault, guarded bool) error {
|
||
if m.ID == "" {
|
||
m.ID = newID("msg")
|
||
}
|
||
m.CreatedAt = m.CreatedAt.UTC()
|
||
insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id)
|
||
SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?`
|
||
// 事务内收口（U3，KTD3/KTD7）：lockConversation（PG 事务级咨询锁；
|
||
// SQLite 单写连接天然串行）后 MAX(seq)+1 分配与 INSERT 同事务，序号
|
||
// 竞态在锁下消失。唯一约束仍留作兜底——冲突只剩事务边界外的极端
|
||
// 竞态（如手工写入），小退避重试即可收敛。
|
||
for attempt := 0; attempt < 8; attempt++ {
|
||
err := s.appendMessageTx(m, requestID, insert, actor, allowImplicitDefault, guarded)
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
if !isUniqueViolation(err) {
|
||
return err
|
||
}
|
||
if requestID != "" {
|
||
existing, lookErr := s.MessageByRequest(string(m.ConversationID), requestID)
|
||
if lookErr == nil {
|
||
*m = existing
|
||
return &DuplicateRequestError{Existing: existing}
|
||
}
|
||
}
|
||
backoff := attempt * attempt
|
||
if backoff > 25 {
|
||
backoff = 25
|
||
}
|
||
time.Sleep(time.Duration(backoff) * time.Millisecond)
|
||
}
|
||
return fmt.Errorf("store: message seq conflict retries exhausted (conversation %s)", m.ConversationID)
|
||
}
|
||
|
||
// appendMessageTx 单事务完成同会话消息追加：会话锁 → 序号分配 → INSERT。
|
||
// 请求幂等依赖 uq_messages_conv_request 唯一索引在事务内暴露冲突。
|
||
func (s *Store) appendMessageTx(m *core.Message, requestID, insert, actor string, allowImplicitDefault, guarded bool) error {
|
||
if guarded {
|
||
return s.withEffectiveAccountWrite(actor, allowImplicitDefault, func(tx *sql.Tx, _ Account) error {
|
||
return s.appendMessageRows(tx, m, requestID, insert, actor)
|
||
})
|
||
}
|
||
tx, err := s.db.Begin()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
if err := s.appendMessageRows(tx, m, requestID, insert, ""); err != nil {
|
||
return err
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
func (s *Store) appendMessageRows(tx *sql.Tx, m *core.Message, requestID, insert, actor string) error {
|
||
if actor != "" {
|
||
var owner string
|
||
if err := tx.QueryRow(s.q(`SELECT owner FROM conversations WHERE id=?`), m.ConversationID).Scan(&owner); errors.Is(err, sql.ErrNoRows) {
|
||
return ErrNotFound
|
||
} else if err != nil {
|
||
return err
|
||
}
|
||
if owner != actor {
|
||
return ErrNotFound
|
||
}
|
||
}
|
||
if err := s.dialect.lockConversation(tx, string(m.ConversationID)); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID); err != nil {
|
||
return err
|
||
}
|
||
_, err := tx.Exec(s.q(`UPDATE conversations SET updated_at=? WHERE id=?`), nowISO(), m.ConversationID)
|
||
return err
|
||
}
|
||
|
||
func (s *Store) ListMessages(conversationID string) ([]core.Message, error) {
|
||
rows, err := s.db.Query(s.q(`SELECT id,conversation_id,run_id,role,parts,created_at,excluded,variant_group,variant_index FROM messages WHERE conversation_id=? ORDER BY seq`), conversationID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := []core.Message{}
|
||
for rows.Next() {
|
||
var m core.Message
|
||
var parts, ca string
|
||
var excluded, index int
|
||
if err := rows.Scan(&m.ID, &m.ConversationID, &m.RunID, &m.Role, &parts, &ca, &excluded, &m.VariantGroup, &index); err != nil {
|
||
return nil, err
|
||
}
|
||
m.Role = core.Role(m.Role)
|
||
m.Parts, _ = unmarshalJSON[[]core.Part](parts)
|
||
m.CreatedAt, _ = parseTime(ca)
|
||
m.Excluded = excluded == 1
|
||
m.VariantIndex = index
|
||
out = append(out, m)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
func (s *Store) LastMessages(conversationID string, n int) ([]core.Message, error) {
|
||
all, err := s.ListMessages(conversationID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(all) > n {
|
||
all = all[len(all)-n:]
|
||
}
|
||
return all, nil
|
||
}
|
||
|
||
// SubagentRunIDs 返回会话里子代理 run 的 id。主对话历史用它排除子 run 消息。
|
||
func (s *Store) SubagentRunIDs(conversationID string) (map[core.ID]bool, error) {
|
||
rows, err := s.db.Query(s.q(`SELECT id FROM runs WHERE conversation_id=? AND kind=?`), conversationID, string(core.RunSubagent))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := map[core.ID]bool{}
|
||
for rows.Next() {
|
||
var id string
|
||
if err := rows.Scan(&id); err != nil {
|
||
return nil, err
|
||
}
|
||
out[core.ID(id)] = true
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// ---- Runs（owner 冗余自会话；GetRun 面向 API 带租户过滤，内部路径用 getRunByID）----
|
||
|
||
func (s *Store) CreateRun(r *core.Run) error {
|
||
if r.ID == "" {
|
||
r.ID = newID("run")
|
||
}
|
||
r.CreatedAt = r.CreatedAt.UTC()
|
||
_, err := s.db.Exec(s.q(`INSERT INTO runs(id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at,instance_id) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`),
|
||
r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, nowISO(), s.instanceID)
|
||
return err
|
||
}
|
||
|
||
func (s *Store) UpdateRun(r *core.Run) error {
|
||
var ended any
|
||
if r.EndedAt != nil {
|
||
ended = nowISO()
|
||
}
|
||
_, err := s.db.Exec(s.q(`UPDATE runs SET status=?, phase=?, usage=?, error=?, ended_at=? WHERE id=?`),
|
||
string(r.Status), string(r.Phase), marshalJSON(r.Usage), marshalJSON(r.Error), ended, r.ID)
|
||
return err
|
||
}
|
||
|
||
const runColumns = `id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at,ended_at,usage,error`
|
||
|
||
func scanRun(scan func(...any) error) (core.Run, error) {
|
||
var r core.Run
|
||
var parent, owner, profile, phase, agent, ended, usage, errJSON *string
|
||
var kind, status string
|
||
var ca string
|
||
if err := scan(&r.ID, &parent, &r.ConversationID, &r.ProjectID, &owner, &kind, &profile, &status, &phase, &agent, &ca, &ended, &usage, &errJSON); err != nil {
|
||
return r, err
|
||
}
|
||
if parent != nil {
|
||
r.ParentRunID = *parent
|
||
}
|
||
if owner != nil && *owner != "" {
|
||
r.Owner = *owner
|
||
}
|
||
if profile != nil {
|
||
r.ProfileID = *profile
|
||
}
|
||
r.Kind = core.RunKind(kind)
|
||
r.Status = core.RunStatus(status)
|
||
if phase != nil {
|
||
r.Phase = core.RunPhase(*phase)
|
||
}
|
||
if agent != nil {
|
||
r.AgentName = *agent
|
||
}
|
||
r.CreatedAt, _ = parseTime(ca)
|
||
if ended != nil {
|
||
t, _ := parseTime(*ended)
|
||
r.EndedAt = &t
|
||
}
|
||
if usage != nil && *usage != "" && *usage != "null" {
|
||
r.Usage, _ = unmarshalJSON[*core.Usage](*usage)
|
||
}
|
||
if errJSON != nil && *errJSON != "" && *errJSON != "null" {
|
||
r.Error, _ = unmarshalJSON[*core.ErrorPart](*errJSON)
|
||
}
|
||
return r, nil
|
||
}
|
||
|
||
// GetRun 面向 API 的租户过滤读取（跨租户 = NotFound）。
|
||
func (s *Store) GetRun(owner, id string) (core.Run, error) {
|
||
r, err := scanRun(func(dst ...any) error {
|
||
return s.db.QueryRow(s.q(`SELECT `+runColumns+` FROM runs WHERE id=? AND owner=?`), id, owner).Scan(dst...)
|
||
})
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return r, ErrNotFound
|
||
}
|
||
return r, err
|
||
}
|
||
|
||
// GetRunByID 内部读取，无租户过滤（runmgr 生命周期管理；API 层必须用 GetRun）。
|
||
func (s *Store) GetRunByID(id string) (core.Run, error) {
|
||
r, err := scanRun(func(dst ...any) error {
|
||
return s.db.QueryRow(s.q(`SELECT `+runColumns+` FROM runs WHERE id=?`), id).Scan(dst...)
|
||
})
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return r, ErrNotFound
|
||
}
|
||
return r, err
|
||
}
|
||
|
||
func (s *Store) ListRunsByConversation(conversationID string) ([]core.Run, error) {
|
||
rows, err := s.db.Query(s.q(`SELECT `+runColumns+` FROM runs WHERE conversation_id=? ORDER BY created_at`), conversationID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := []core.Run{}
|
||
for rows.Next() {
|
||
r, err := scanRun(rows.Scan)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, r)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// ChildRuns 返回直接子 run（任务树）。
|
||
func (s *Store) ChildRuns(runID string) ([]core.Run, error) {
|
||
rows, err := s.db.Query(s.q(`SELECT `+runColumns+` FROM runs WHERE parent_run_id=? ORDER BY created_at`), runID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := []core.Run{}
|
||
for rows.Next() {
|
||
r, err := scanRun(rows.Scan)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, r)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// ---- 事件（会话内连续 seq 游标权威；批量落盘降低写放大）----
|
||
|
||
// AppendEvents 在单事务内追加一批事件并分配连续 seq（base+1..base+n），
|
||
// 成功后回填 evs[i].Seq。事务内锁定会话序列：PG 用事务级咨询锁（跨实例安全），
|
||
// SQLite 单写连接天然串行。空批次为空操作。
|
||
func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int64, error) {
|
||
return s.appendEvents(conversationID, evs, "", "")
|
||
}
|
||
|
||
func (s *Store) appendEvents(conversationID string, evs []core.StreamEvent, owner, producer string) (int64, error) {
|
||
if len(evs) == 0 {
|
||
return 0, nil
|
||
}
|
||
tx, err := s.db.Begin()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
defer tx.Rollback()
|
||
base, err := s.appendEventsInTx(tx, conversationID, evs, owner, producer)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return 0, err
|
||
}
|
||
assignEventSequences(evs, base)
|
||
return base, nil
|
||
}
|
||
|
||
// AppendEventsAccountChecked is the final persistence boundary for an
|
||
// actor-attributed card/surface event. The actor account lock precedes the
|
||
// channel sequence lock and event insert.
|
||
func (s *Store) AppendEventsAccountChecked(actor, conversationID string, evs []core.StreamEvent, allowImplicitDefault bool) (int64, error) {
|
||
if len(evs) == 0 {
|
||
return 0, nil
|
||
}
|
||
var base int64
|
||
err := s.withEffectiveAccountWrite(actor, allowImplicitDefault, func(tx *sql.Tx, _ Account) error {
|
||
if projectID, projectChannel := core.ProjectFromChannel(core.ID(conversationID)); projectChannel {
|
||
var one int
|
||
if err := tx.QueryRow(s.q(`SELECT 1 FROM projects WHERE id=?`), projectID).Scan(&one); errors.Is(err, sql.ErrNoRows) {
|
||
return ErrNotFound
|
||
} else if err != nil {
|
||
return err
|
||
}
|
||
} else {
|
||
var owner string
|
||
if err := tx.QueryRow(s.q(`SELECT owner FROM conversations WHERE id=?`), conversationID).Scan(&owner); errors.Is(err, sql.ErrNoRows) {
|
||
return ErrNotFound
|
||
} else if err != nil {
|
||
return err
|
||
}
|
||
if owner != actor {
|
||
return ErrNotFound
|
||
}
|
||
}
|
||
var err error
|
||
base, err = s.appendEventsInTx(tx, conversationID, evs, "", "")
|
||
return err
|
||
})
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
assignEventSequences(evs, base)
|
||
return base, nil
|
||
}
|
||
|
||
// AppendEventsForEffectiveChannel resolves an internal run event's account
|
||
// from the persisted channel before entering the final guarded append. Card
|
||
// and Surface APIs use their explicit actor-aware emitter instead.
|
||
func (s *Store) AppendEventsForEffectiveChannel(conversationID string, evs []core.StreamEvent, allowImplicitDefault bool) (int64, error) {
|
||
var owner string
|
||
if projectID, projectChannel := core.ProjectFromChannel(core.ID(conversationID)); projectChannel {
|
||
if err := s.db.QueryRow(s.q(`SELECT owner FROM projects WHERE id=?`), projectID).Scan(&owner); errors.Is(err, sql.ErrNoRows) {
|
||
return 0, ErrNotFound
|
||
} else if err != nil {
|
||
return 0, err
|
||
}
|
||
} else {
|
||
if err := s.db.QueryRow(s.q(`SELECT owner FROM conversations WHERE id=?`), conversationID).Scan(&owner); errors.Is(err, sql.ErrNoRows) {
|
||
return 0, ErrNotFound
|
||
} else if err != nil {
|
||
return 0, err
|
||
}
|
||
}
|
||
return s.AppendEventsAccountChecked(owner, conversationID, evs, allowImplicitDefault)
|
||
}
|
||
|
||
func assignEventSequences(evs []core.StreamEvent, base int64) {
|
||
for i := range evs {
|
||
evs[i].Seq = base + int64(i) + 1
|
||
}
|
||
}
|
||
|
||
func (s *Store) appendEventsInTx(tx *sql.Tx, conversationID string, evs []core.StreamEvent, owner, producer string) (int64, error) {
|
||
if err := s.dialect.lockConversation(tx, conversationID); err != nil {
|
||
return 0, err
|
||
}
|
||
var base sql.NullInt64
|
||
if err := tx.QueryRow(s.q(`SELECT COALESCE(MAX(seq),0) FROM events WHERE conversation_id=?`), conversationID).Scan(&base); err != nil {
|
||
return 0, err
|
||
}
|
||
var sb strings.Builder
|
||
sb.WriteString(`INSERT INTO events(conversation_id,seq,run_id,ts,v,type,payload) VALUES `)
|
||
args := make([]any, 0, len(evs)*7)
|
||
for i := range evs {
|
||
if i > 0 {
|
||
sb.WriteString(",")
|
||
}
|
||
sb.WriteString("(?,?,?,?,?,?,?)")
|
||
var payload any
|
||
if len(evs[i].Payload) > 0 {
|
||
payload = []byte(evs[i].Payload)
|
||
}
|
||
args = append(args, conversationID, base.Int64+int64(i)+1, evs[i].RunID, evs[i].TS, evs[i].V, string(evs[i].Type), payload)
|
||
}
|
||
if _, err := tx.Exec(s.q(sb.String()), args...); err != nil {
|
||
return 0, err
|
||
}
|
||
if producer != "" {
|
||
for i, e := range evs {
|
||
if err := s.saveProjectEventProof(tx, conversationID, base.Int64+int64(i)+1, e, owner, producer); err != nil {
|
||
return 0, err
|
||
}
|
||
}
|
||
}
|
||
// KTD6：事件提交与 NOTIFY 同事务（outbox 语义的前半——通知不早于事件
|
||
// 可见；后半的命令 outbox 走 EnqueueCommandInTx 由批准路径使用）。
|
||
if s.dialect.name() == "postgres" {
|
||
if _, err := tx.Exec(`SELECT pg_notify($1, $2)`, "agent_events",
|
||
conversationID+"|"+strconv.FormatInt(base.Int64+int64(len(evs)), 10)); err != nil {
|
||
return 0, err
|
||
}
|
||
}
|
||
return base.Int64, nil
|
||
}
|
||
|
||
func (s *Store) EventsAfter(conversationID string, since int64, limit int) ([]core.StreamEvent, error) {
|
||
rows, err := s.db.Query(s.q(`SELECT seq,run_id,ts,v,type,payload FROM events WHERE conversation_id=? AND seq>? ORDER BY seq LIMIT ?`),
|
||
conversationID, since, limit)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := []core.StreamEvent{}
|
||
for rows.Next() {
|
||
var e core.StreamEvent
|
||
var typ string
|
||
var payload []byte
|
||
if err := rows.Scan(&e.Seq, &e.RunID, &e.TS, &e.V, &typ, &payload); err != nil {
|
||
return nil, err
|
||
}
|
||
e.ConversationID = conversationID
|
||
// 项目流重放（D50）：信封盖章与实时路径同源（伪会话键反解）。
|
||
if pid, ok := core.ProjectFromChannel(core.ID(conversationID)); ok {
|
||
e.ProjectID = pid
|
||
}
|
||
e.Type = core.StreamEventType(typ)
|
||
if len(payload) > 0 {
|
||
e.Payload = payload
|
||
}
|
||
out = append(out, e)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// RunEventsAfter reads a run's persisted stream plus its direct children's
|
||
// collaboration status events in conversation seq order. Raw child output is
|
||
// intentionally outside the parent's previously visible process stream.
|
||
// The API checks conversation and run ownership before calling this method.
|
||
func (s *Store) RunEventsAfter(conversationID, runID string, since int64, limit int) ([]core.StreamEvent, error) {
|
||
rows, err := s.db.Query(s.q(`SELECT seq,run_id,ts,v,type,payload FROM events
|
||
WHERE conversation_id=? AND seq>? AND (
|
||
run_id=? OR (run_id IN (SELECT id FROM runs WHERE parent_run_id=? AND conversation_id=?)
|
||
AND type IN ('agent.progress','agent.result'))
|
||
) ORDER BY seq LIMIT ?`), conversationID, since, runID, runID, conversationID, limit)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
out := []core.StreamEvent{}
|
||
for rows.Next() {
|
||
var e core.StreamEvent
|
||
var typ string
|
||
var payload []byte
|
||
if err := rows.Scan(&e.Seq, &e.RunID, &e.TS, &e.V, &typ, &payload); err != nil {
|
||
return nil, err
|
||
}
|
||
e.ConversationID = conversationID
|
||
e.Type = core.StreamEventType(typ)
|
||
if len(payload) > 0 {
|
||
e.Payload = payload
|
||
}
|
||
out = append(out, e)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// EventAt 读取单个事件（PG 档 NOTIFY 扇出的按序号取回路径）。
|
||
func (s *Store) EventAt(conversationID string, seq int64) (core.StreamEvent, error) {
|
||
var e core.StreamEvent
|
||
var typ string
|
||
var payload []byte
|
||
err := s.db.QueryRow(s.q(`SELECT seq,run_id,ts,v,type,payload FROM events WHERE conversation_id=? AND seq=?`), conversationID, seq).
|
||
Scan(&e.Seq, &e.RunID, &e.TS, &e.V, &typ, &payload)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return e, ErrNotFound
|
||
}
|
||
if err != nil {
|
||
return e, err
|
||
}
|
||
e.ConversationID = conversationID
|
||
e.Type = core.StreamEventType(typ)
|
||
if len(payload) > 0 {
|
||
e.Payload = payload
|
||
}
|
||
return e, nil
|
||
}
|
||
|
||
func (s *Store) LastEventSeq(conversationID string) (int64, error) {
|
||
var seq sql.NullInt64
|
||
err := s.db.QueryRow(s.q(`SELECT MAX(seq) FROM events WHERE conversation_id=?`), conversationID).Scan(&seq)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return seq.Int64, nil
|
||
}
|
||
|
||
// FirstRunEventSeq 返回这个 run 在会话事件流里的第一条 seq。
|
||
// 没有事件时 ok 为 false。调用方用 seq-1 作为重放游标。
|
||
func (s *Store) FirstRunEventSeq(conversationID, runID string) (seq int64, ok bool, err error) {
|
||
var got sql.NullInt64
|
||
err = s.db.QueryRow(s.q(`SELECT MIN(seq) FROM events WHERE conversation_id=? AND run_id=?`), conversationID, runID).Scan(&got)
|
||
if err != nil || !got.Valid {
|
||
return 0, false, err
|
||
}
|
||
return got.Int64, true, nil
|
||
}
|
||
|
||
func newID(kind string) string {
|
||
b := make([]byte, 6)
|
||
rand.Read(b)
|
||
return kind + "_" + hex.EncodeToString(b)
|
||
}
|
||
|
||
func parseTime(s string) (time.Time, error) {
|
||
return time.Parse(time.RFC3339Nano, s)
|
||
}
|
||
|
||
// NewID 生成带前缀的随机 ID（导出供 runmgr/server 使用）。
|
||
func NewID(kind string) string { return newID(kind) }
|