feat(agent): 运行详情分页与增量读取契约 (#600)
Co-authored-by: ddcat666 <26564566+ddcat666@users.noreply.github.com>
This commit is contained in:
@@ -88,6 +88,21 @@ type CloudAgentRun struct {
|
||||
SpentCredits float64 `json:"spentCredits"`
|
||||
Step int `json:"step"`
|
||||
ActiveMessage map[string]string `json:"activeMessage,omitempty"`
|
||||
// 事件已全量落库,运行详情只返回一页,因此必须把"这一页在整条日志里的位置"说清楚:
|
||||
// EventSeqBase 是本次返回的首条事件之前已入库的条数(不变量 events[i].seq ==
|
||||
// eventSeqBase + i + 1),EventCount 是该运行累计事件条数,LatestSeq 可直接当作下次
|
||||
// 增量读取的 sinceSeq,EventsTruncated 表示还有更早的记录没随本次返回。
|
||||
EventSeqBase int `json:"eventSeqBase"`
|
||||
EventCount int `json:"eventCount"`
|
||||
LatestSeq int `json:"latestSeq"`
|
||||
EventsTruncated bool `json:"eventsTruncated"`
|
||||
}
|
||||
|
||||
// CloudAgentRunViewOptions 是运行详情的读取选项:SinceSeq 只取该序号之后的增量,
|
||||
// EventLimit 覆盖默认页大小。零值即默认视图(尾部一窗)。
|
||||
type CloudAgentRunViewOptions struct {
|
||||
SinceSeq int
|
||||
EventLimit int
|
||||
}
|
||||
|
||||
func validateCloudAgentRequest(req *CloudAgentRequest) error {
|
||||
@@ -267,7 +282,7 @@ func cloudAgentRunTerminal(status string) bool {
|
||||
return status == "completed" || status == "failed" || status == "cancelled" || status == "rejected"
|
||||
}
|
||||
|
||||
func (s *Service) CloudAgentRun(userID, id string) (*CloudAgentRun, error) {
|
||||
func (s *Service) CloudAgentRun(userID, id string, options ...CloudAgentRunViewOptions) (*CloudAgentRun, error) {
|
||||
task, state, err := s.cloudAgentTask(userID, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -284,13 +299,13 @@ func (s *Service) CloudAgentRun(userID, id string) (*CloudAgentRun, error) {
|
||||
} else if lookupErr != nil {
|
||||
return nil, lookupErr
|
||||
}
|
||||
return s.cloudAgentExecutionOutput(task, state)
|
||||
return s.cloudAgentExecutionOutput(task, state, options...)
|
||||
}
|
||||
|
||||
// CloudAgentRunIfChanged keeps idle event streams on a small indexed read.
|
||||
// The persisted revision, not a process-local notification, is authoritative
|
||||
// across instances and after missed/disconnected notifications.
|
||||
func (s *Service) CloudAgentRunIfChanged(userID, id string, revision int64) (*CloudAgentRun, error) {
|
||||
func (s *Service) CloudAgentRunIfChanged(userID, id string, revision int64, options ...CloudAgentRunViewOptions) (*CloudAgentRun, error) {
|
||||
current, err := s.repo.CloudAgentRevision(userID, id)
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, kernel.NotFound("Agent 运行不存在")
|
||||
@@ -301,7 +316,7 @@ func (s *Service) CloudAgentRunIfChanged(userID, id string, revision int64) (*Cl
|
||||
if current == revision {
|
||||
return nil, nil
|
||||
}
|
||||
return s.CloudAgentRun(userID, id)
|
||||
return s.CloudAgentRun(userID, id, options...)
|
||||
}
|
||||
|
||||
// CreateCloudAgentRun validates every capability before admission. The task PK
|
||||
@@ -364,7 +379,9 @@ func (s *Service) CreateCloudAgentRun(userID string, req CloudAgentRequest, pare
|
||||
if err := s.advanceCloudAgentByID(userID, parentID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
parentRun, err := s.CloudAgentRun(userID, parentID)
|
||||
// 续轮收束要读上一轮**全部**事件(运行详情默认只返回尾部一窗):长会话一旦被截断,
|
||||
// 新轮就看不到上一轮改过哪些节点、提交过哪些任务,表现为"忘了自己做过什么"。
|
||||
parentRun, err := s.CloudAgentRun(userID, parentID, CloudAgentRunViewOptions{EventLimit: cloudAgentContinuationEventLimit})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -17,6 +17,10 @@ const (
|
||||
// cloudAgentContinuationFrameBytes 是交接帧的硬上限:超了就按"最近的改动优先"裁剪,
|
||||
// 并把裁掉的组数记进 omittedCanvasChanges,绝不把整份历史顶过硬闸。
|
||||
cloudAgentContinuationFrameBytes = 8 << 10
|
||||
// cloudAgentContinuationEventLimit 是续轮读取上一轮事件的上限。运行详情默认只返回尾部
|
||||
// 一窗,而收束要覆盖上一轮**全部**真实改动(画布改动、已提交任务、助手原话),
|
||||
// 被窗口截断就会表现为"新一轮忘了上一轮做过什么"。
|
||||
cloudAgentContinuationEventLimit = 1000
|
||||
)
|
||||
|
||||
type cloudAgentContinuationChange struct {
|
||||
|
||||
148
backend/internal/app/cloud_agent_event_store.go
Normal file
148
backend/internal/app/cloud_agent_event_store.go
Normal file
@@ -0,0 +1,148 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
)
|
||||
|
||||
const (
|
||||
// cloudAgentRunEventDeltaLimit 是增量读取(sinceSeq)单次返回的条数上限,
|
||||
// 也是 HTTP 上 eventLimit 允许的最大值(越界的请求由 handler 拒绝)。
|
||||
// 内部调用方不受它约束:续轮收束按 cloudAgentContinuationEventLimit 显式取更大的页。
|
||||
cloudAgentRunEventDeltaLimit = 500
|
||||
// cloudAgentRunEventPageLimit 是运行详情默认返回的事件条数:等于内存里的尾部窗口,
|
||||
// 因此默认读取不额外查事件表;要更早的记录由客户端通过 eventLimit 显式索取。
|
||||
cloudAgentRunEventPageLimit = repository.CloudAgentJournalWindow
|
||||
// cloudAgentEventWindowSanityLimit 是内存事件窗口的健全上限(窗口本身加一次转移
|
||||
// 新产生的事件),超过它说明窗口没有按 CloudAgentJournalWindow 载入。
|
||||
cloudAgentEventWindowSanityLimit = 512
|
||||
)
|
||||
|
||||
// cloudAgentRunEventsForView 组装运行详情要返回的事件。
|
||||
//
|
||||
// 事件全量在 cloud_agent_event_records(一行一条 EventJSON,主键 run_id + sequence),
|
||||
// 内存里只有最近一窗(repository.CloudAgentJournalWindow),所以两条读路径都要显式
|
||||
// 说明"这一页在整条日志里的位置":
|
||||
//
|
||||
// - sinceSeq > 0:只返回该序号之后的增量(SSE 断线重连 / 滚动拉取),走
|
||||
// `sequence > ? ORDER BY sequence LIMIT ?` 的 keyset 读,不载入更早的行。
|
||||
// - sinceSeq == 0:返回最近 cloudAgentRunEventPageLimit 条;窗口不够时(eventLimit
|
||||
// 大于窗口)再从事件表往前补一页,仍然只按 seq 读。
|
||||
//
|
||||
// 事件表里没有该运行的行时(升级前的旧检查点把事件放在 state_json 里),退回内存窗口,
|
||||
// 保证老运行仍可读。
|
||||
func (s *Service) cloudAgentRunEventsForView(userID string, run *model.CloudAgentExecution, state *cloudAgentRuntime, sinceSeq, limit int) []CloudAgentEvent {
|
||||
if sinceSeq > 0 {
|
||||
pageLimit := cloudAgentRunEventDeltaLimit
|
||||
if limit > 0 && limit < pageLimit {
|
||||
pageLimit = limit
|
||||
}
|
||||
rows, err := s.repo.CloudAgentEventRecords(userID, run.ID, sinceSeq, pageLimit)
|
||||
if err != nil {
|
||||
log.Printf("agent event delta %s: %v", run.ID, err)
|
||||
rows = nil
|
||||
}
|
||||
events := make([]CloudAgentEvent, 0, pageLimit)
|
||||
for _, row := range rows {
|
||||
events = append(events, cloudAgentEventFromRecord(row))
|
||||
}
|
||||
// 数据库页已达到上限时不能再拼接内存窗口,否则可能超过请求的页大小,
|
||||
// 或在表中间隔着未返回的历史记录直接跳到窗口尾部,制造序号缺口。
|
||||
if len(events) == pageLimit {
|
||||
return events
|
||||
}
|
||||
last := sinceSeq
|
||||
if len(rows) > 0 {
|
||||
last = rows[len(rows)-1].Sequence
|
||||
}
|
||||
// 表未填满这一页时,内存里可能还有本次转移尚未落库的事件,补在末尾。
|
||||
for _, event := range state.Events {
|
||||
if event.Seq > last {
|
||||
events = append(events, event)
|
||||
if len(events) == pageLimit {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(events) > 0 {
|
||||
return events
|
||||
}
|
||||
// 表与窗口都没有更新的内容:按游标过滤内存窗口(旧检查点的运行走这里)。
|
||||
pending := make([]CloudAgentEvent, 0, len(state.Events))
|
||||
for _, event := range state.Events {
|
||||
if event.Seq > sinceSeq {
|
||||
pending = append(pending, event)
|
||||
if len(pending) == pageLimit {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return pending
|
||||
}
|
||||
tail := state.Events
|
||||
if limit <= 0 {
|
||||
limit = cloudAgentRunEventPageLimit
|
||||
}
|
||||
if len(tail) >= limit {
|
||||
return append([]CloudAgentEvent(nil), tail[len(tail)-limit:]...)
|
||||
}
|
||||
// 窗口已经是全量(水位 0):没有更早的记录可补,直接返回。
|
||||
if state.EventSeqBase == 0 {
|
||||
return append([]CloudAgentEvent(nil), tail...)
|
||||
}
|
||||
rows, err := s.repo.CloudAgentEventRecordsBefore(userID, run.ID, state.EventSeqBase+1, limit-len(tail))
|
||||
if err != nil {
|
||||
log.Printf("agent event page %s: %v", run.ID, err)
|
||||
return append([]CloudAgentEvent(nil), tail...)
|
||||
}
|
||||
events := make([]CloudAgentEvent, 0, len(rows)+len(tail))
|
||||
for _, row := range rows {
|
||||
events = append(events, cloudAgentEventFromRecord(row))
|
||||
}
|
||||
return append(events, tail...)
|
||||
}
|
||||
|
||||
// cloudAgentEventFromRecord 把事件行解析成对外契约结构。
|
||||
func cloudAgentEventFromRecord(row model.CloudAgentEventRecord) CloudAgentEvent {
|
||||
var event CloudAgentEvent
|
||||
if err := json.Unmarshal([]byte(row.EventJSON), &event); err != nil || event.EventID == "" {
|
||||
// 行存在但解不出来:仍按行给出身份与占位类型,避免整页读取失败。
|
||||
// 这条事件不代表运行的真实动作,客户端应以服务端画布与任务状态为准。
|
||||
return CloudAgentEvent{
|
||||
EventID: fmt.Sprintf("%s:%d", row.RunID, row.Sequence), RunID: row.RunID, Seq: row.Sequence,
|
||||
Type: "unreadable_event", Payload: map[string]any{"text": "该事件记录无法解析,请以服务端画布与任务状态为准"}, CreatedAt: row.CreatedAt,
|
||||
}
|
||||
}
|
||||
if event.RunID == "" {
|
||||
event.RunID = row.RunID
|
||||
}
|
||||
if event.Seq == 0 {
|
||||
event.Seq = row.Sequence
|
||||
}
|
||||
if event.CreatedAt.IsZero() {
|
||||
event.CreatedAt = row.CreatedAt
|
||||
}
|
||||
return event
|
||||
}
|
||||
|
||||
// cloudAgentRunEventCount 给出"这个运行一共产生过多少事件":以事件表条数为准,再补上
|
||||
// 本次转移还没落库的部分(内存里 seq 高于水位的窗口)与检查点水位,取三者最大值。
|
||||
func (s *Service) cloudAgentRunEventCount(userID string, run *model.CloudAgentExecution, state *cloudAgentRuntime) int {
|
||||
stored, err := s.repo.CloudAgentEventRecordCount(userID, run.ID)
|
||||
if err != nil {
|
||||
log.Printf("agent event count %s: %v", run.ID, err)
|
||||
stored = 0
|
||||
}
|
||||
total := int(stored)
|
||||
if pending := state.EventSeqBase + len(state.Events); pending > total {
|
||||
total = pending
|
||||
}
|
||||
if run != nil && run.EventCount > total {
|
||||
total = run.EventCount
|
||||
}
|
||||
return total
|
||||
}
|
||||
320
backend/internal/app/cloud_agent_event_store_test.go
Normal file
320
backend/internal/app/cloud_agent_event_store_test.go
Normal file
@@ -0,0 +1,320 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// 运行详情的事件分页契约。事件全量在 cloud_agent_event_records(一行一条 EventJSON,
|
||||
// 主键 run_id + sequence),内存只保留最近一窗,因此读路径必须自己说明"这一页在整条
|
||||
// 日志里的位置"。本用例守住四件事:
|
||||
// - 载入(hydrate + decode)只取一窗,水位落在窗口之前,窗口末条等于 EventCount;
|
||||
// - 默认视图是尾部一窗,events[i].Seq == EventSeqBase + i + 1 成立;
|
||||
// - sinceSeq 只返回增量,eventLimit 决定页大小(大于窗口时从事件表往前补);
|
||||
// - eventCount / latestSeq / eventsTruncated 三个字段互相自洽。
|
||||
func TestCloudAgentRunDetailPagesJournalBySeq(t *testing.T) {
|
||||
s, _, root := reliableAgentRoot(t)
|
||||
// 事件数刻意跨过内存窗口与 HTTP 单次上限,才能同时验证"尾部窗口"、"从表里往前补页"
|
||||
// 和"内部调用方(续轮收束)可以要更大的页"三条路径。
|
||||
total := cloudAgentRunEventDeltaLimit + repository.CloudAgentJournalWindow*2
|
||||
growAgentJournal(t, s, root.ID, total)
|
||||
stored, err := s.repo.CloudAgentEventRecordCount("user", root.ID)
|
||||
if err != nil || int(stored) != total {
|
||||
t.Fatalf("入库条数 = %d(err=%v),期望 %d", stored, err, total)
|
||||
}
|
||||
|
||||
persisted, err := s.repo.CloudAgent("user", root.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if persisted.EventCount != total {
|
||||
t.Fatalf("事件水位 = %d,期望 %d", persisted.EventCount, total)
|
||||
}
|
||||
rebuilt, err := cloudAgentDecode(persisted)
|
||||
if err != nil {
|
||||
t.Fatalf("重建失败: %v", err)
|
||||
}
|
||||
if len(rebuilt.Events) != repository.CloudAgentJournalWindow {
|
||||
t.Fatalf("载入窗口 = %d 条,期望 %d", len(rebuilt.Events), repository.CloudAgentJournalWindow)
|
||||
}
|
||||
if rebuilt.EventSeqBase != total-repository.CloudAgentJournalWindow {
|
||||
t.Fatalf("窗口水位 = %d,期望 %d", rebuilt.EventSeqBase, total-repository.CloudAgentJournalWindow)
|
||||
}
|
||||
if rebuilt.Events[len(rebuilt.Events)-1].Seq != total {
|
||||
t.Fatalf("窗口末条序号 = %d,期望 %d", rebuilt.Events[len(rebuilt.Events)-1].Seq, total)
|
||||
}
|
||||
|
||||
// 默认视图:尾部一窗,并且如实报告"还有更早的记录"。
|
||||
view, err := s.CloudAgentRun("user", root.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, view, total-repository.CloudAgentJournalWindow, total, true)
|
||||
if len(view.Events) != cloudAgentRunEventPageLimit {
|
||||
t.Fatalf("默认页 = %d 条,期望 %d", len(view.Events), cloudAgentRunEventPageLimit)
|
||||
}
|
||||
|
||||
// eventLimit 是页大小而不是忽略:小于窗口时只返回尾部这么多条。
|
||||
narrowed, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{EventLimit: 10})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, narrowed, total-10, total, true)
|
||||
if len(narrowed.Events) != 10 {
|
||||
t.Fatalf("eventLimit=10 返回 %d 条", len(narrowed.Events))
|
||||
}
|
||||
|
||||
// eventLimit 大于窗口:必须从事件表往前补齐,而不是只给窗口里那 40 条。
|
||||
deeper, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{EventLimit: repository.CloudAgentJournalWindow + 10})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, deeper, total-repository.CloudAgentJournalWindow-10, total, true)
|
||||
if len(deeper.Events) != repository.CloudAgentJournalWindow+10 {
|
||||
t.Fatalf("补齐后页大小 = %d,期望 %d", len(deeper.Events), repository.CloudAgentJournalWindow+10)
|
||||
}
|
||||
|
||||
// HTTP 单次上限(500)刚好覆盖不到整条日志:页对齐到 500,并如实标记还有更早记录。
|
||||
capped, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{EventLimit: cloudAgentRunEventDeltaLimit})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, capped, total-cloudAgentRunEventDeltaLimit, total, true)
|
||||
if len(capped.Events) != cloudAgentRunEventDeltaLimit {
|
||||
t.Fatalf("上限页 = %d 条,期望 %d", len(capped.Events), cloudAgentRunEventDeltaLimit)
|
||||
}
|
||||
|
||||
// 内部调用方(续轮收束)不受 HTTP 上限约束:更大的页必须能拿到整条日志,
|
||||
// 否则交接帧会因为窗口而漏掉上一轮的真实改动。
|
||||
full, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{EventLimit: cloudAgentContinuationEventLimit})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, full, 0, total, false)
|
||||
if len(full.Events) != total {
|
||||
t.Fatalf("全量页 = %d 条,期望 %d", len(full.Events), total)
|
||||
}
|
||||
|
||||
// 增量:sinceSeq 之后只返回更新的部分,且不再声称截断(客户端自己拿着游标)。
|
||||
delta, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{SinceSeq: total - 2})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, delta, total-2, total, false)
|
||||
if len(delta.Events) != 2 || delta.Events[0].Seq != total-1 {
|
||||
t.Fatalf("增量视图不对:%+v", delta.Events)
|
||||
}
|
||||
|
||||
// 增量页也必须尊重 eventLimit;不能先取满默认上限,再把内存尾窗追加到页尾。
|
||||
limitedDelta, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{SinceSeq: 1, EventLimit: 10})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(limitedDelta.Events) != 10 || limitedDelta.Events[0].Seq != 2 || limitedDelta.Events[9].Seq != 11 {
|
||||
t.Fatalf("受限增量页不连续或超出上限:%+v", limitedDelta.Events)
|
||||
}
|
||||
|
||||
// 跨窗口边界:游标落在窗口之前,表与窗口要共同补齐且序号连续。
|
||||
crossing, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{SinceSeq: rebuilt.EventSeqBase - 2})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, crossing, rebuilt.EventSeqBase-2, total, false)
|
||||
if len(crossing.Events) != repository.CloudAgentJournalWindow+2 || crossing.Events[0].Seq != rebuilt.EventSeqBase-1 {
|
||||
t.Fatalf("跨边界增量不对:%d 条,首条 %d", len(crossing.Events), crossing.Events[0].Seq)
|
||||
}
|
||||
|
||||
// 游标已经追平(甚至越界):返回空页而不是报错,客户端据此停止轮询。
|
||||
caughtUp, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{SinceSeq: total})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, caughtUp, total-repository.CloudAgentJournalWindow, total, false)
|
||||
if len(caughtUp.Events) != 0 || caughtUp.LatestSeq != 0 {
|
||||
t.Fatalf("追平后仍返回事件:%+v", caughtUp.Events)
|
||||
}
|
||||
beyond, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{SinceSeq: total + 100})
|
||||
if err != nil || len(beyond.Events) != 0 {
|
||||
t.Fatalf("越界游标应返回空页:events=%d err=%v", len(beyond.Events), err)
|
||||
}
|
||||
|
||||
// 追加一条:增量视图只看得到它,latestSeq 就是下一次的游标。
|
||||
appended, err := s.repo.CloudAgent("user", root.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
grown, err := cloudAgentDecode(appended)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
grown.event(root.ID, "assistant_message", map[string]any{"text": "追加一条"})
|
||||
if err := s.repo.MutateCloudAgent("user", root.ID, appended.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error {
|
||||
return cloudAgentSave(current, &grown)
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
next, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{SinceSeq: total})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, next, total, total+1, false)
|
||||
if len(next.Events) != 1 || next.Events[0].Seq != total+1 {
|
||||
t.Fatalf("追加后的增量不对:%+v", next.Events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCloudAgentRunPaginationFieldsAreStableJSON(t *testing.T) {
|
||||
raw, err := json.Marshal(CloudAgentRun{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var encoded map[string]any
|
||||
if err := json.Unmarshal(raw, &encoded); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, field := range []string{"eventSeqBase", "eventCount", "latestSeq", "eventsTruncated"} {
|
||||
if _, ok := encoded[field]; !ok {
|
||||
t.Fatalf("分页字段 %q 在零值响应中缺失:%s", field, raw)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// growAgentJournal 按真实保存路径把事件追加到 total 条。一次转移不可能产生几百条事件
|
||||
// (超过 cloudAgentEventWindowSanityLimit 会被判"窗口异常"),因此分多轮追加:每轮重新
|
||||
// 载入(窗口右移)后再写,顺带覆盖"窗口左移之后继续追加,历史不被截断"。
|
||||
func growAgentJournal(t *testing.T, s *Service, runID string, total int) {
|
||||
t.Helper()
|
||||
for {
|
||||
run, err := s.repo.CloudAgent("user", runID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
state, err := cloudAgentDecode(run)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
missing := total - (state.EventSeqBase + len(state.Events))
|
||||
if missing <= 0 {
|
||||
return
|
||||
}
|
||||
for index := 0; index < min(missing, cloudAgentEventWindowSanityLimit/2); index++ {
|
||||
state.event(runID, "tool_completed", map[string]any{"toolName": "canvas_get_state", "text": strings.Repeat("读", 20)})
|
||||
}
|
||||
if err := s.repo.MutateCloudAgent("user", runID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error {
|
||||
return cloudAgentSave(current, &state)
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// assertAgentEventPage 校验分页契约本身:条数、水位、序号不变量、latestSeq 与截断标记。
|
||||
func assertAgentEventPage(t *testing.T, run *CloudAgentRun, wantBase, wantCount int, wantTruncated bool) {
|
||||
t.Helper()
|
||||
if run.EventSeqBase != wantBase {
|
||||
t.Fatalf("eventSeqBase = %d,期望 %d", run.EventSeqBase, wantBase)
|
||||
}
|
||||
if run.EventCount != wantCount {
|
||||
t.Fatalf("eventCount = %d,期望 %d", run.EventCount, wantCount)
|
||||
}
|
||||
if run.EventsTruncated != wantTruncated {
|
||||
t.Fatalf("eventsTruncated = %v,期望 %v", run.EventsTruncated, wantTruncated)
|
||||
}
|
||||
for index, event := range run.Events {
|
||||
if event.Seq != run.EventSeqBase+index+1 {
|
||||
t.Fatalf("分页不变量被破坏:events[%d].seq = %d,期望 eventSeqBase(%d)+%d+1", index, event.Seq, run.EventSeqBase, index)
|
||||
}
|
||||
}
|
||||
if len(run.Events) > 0 && run.LatestSeq != run.Events[len(run.Events)-1].Seq {
|
||||
t.Fatalf("latestSeq = %d,期望末条 %d", run.LatestSeq, run.Events[len(run.Events)-1].Seq)
|
||||
}
|
||||
// 截断标记只表示"水位之前还有记录",不能凭空出现。
|
||||
if run.EventsTruncated && run.EventSeqBase <= 0 {
|
||||
t.Fatalf("eventsTruncated 与 eventSeqBase 不自洽:base=%d", run.EventSeqBase)
|
||||
}
|
||||
}
|
||||
|
||||
// 事件表里没有该运行的行时(升级前的旧检查点把事件放在 state_json 里),运行详情必须
|
||||
// 仍能读到这些事件,且水位为 0(它们就是全部)。
|
||||
func TestCloudAgentRunDetailFallsBackToLegacyStateEvents(t *testing.T) {
|
||||
s, db, root := reliableAgentRoot(t)
|
||||
run, err := s.repo.CloudAgent("user", root.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
state, err := cloudAgentDecode(run)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for index := 0; index < 5; index++ {
|
||||
state.event(root.ID, "assistant_message", map[string]any{"text": "legacy"})
|
||||
}
|
||||
legacyJSON, err := json.Marshal(state)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 还原成升级前的形态:事件在 state_json 里、事件表为空、水位为 0。
|
||||
if err := db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Where("run_id = ?", root.ID).Delete(&model.CloudAgentEventRecord{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Model(&model.CloudAgentExecution{}).Where("id = ?", root.ID).Updates(map[string]any{
|
||||
"checkpoint_version": 0, "event_count": 0, "message_count": 0, "state_json": string(legacyJSON),
|
||||
}).Error
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
view, err := s.CloudAgentRun("user", root.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertAgentEventPage(t, view, 0, len(state.Events), false)
|
||||
if len(view.Events) != len(state.Events) {
|
||||
t.Fatalf("旧检查点事件回退失败:%d 条,期望 %d", len(view.Events), len(state.Events))
|
||||
}
|
||||
// 增量也要能从内存窗口过滤出来(表里没有行)。
|
||||
cursor := view.Events[1].Seq
|
||||
delta, err := s.CloudAgentRun("user", root.ID, CloudAgentRunViewOptions{SinceSeq: cursor})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(delta.Events) != len(state.Events)-2 || delta.Events[0].Seq != cursor+1 {
|
||||
t.Fatalf("旧检查点增量回退失败:cursor=%d events=%+v", cursor, delta.Events)
|
||||
}
|
||||
}
|
||||
|
||||
// 事件行缺失但水位不为 0:不能把"一行都没读到"当成空日志,否则运行详情会给出
|
||||
// 一个水位虚高、事件却为空的页,客户端会以为历史已经被清理。
|
||||
func TestCloudAgentDecodeRejectsMissingJournalRows(t *testing.T) {
|
||||
s, db, root := reliableAgentRoot(t)
|
||||
run, err := s.repo.CloudAgent("user", root.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
state, err := cloudAgentDecode(run)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
state.event(root.ID, "tool_completed", map[string]any{"toolName": "canvas_get_state"})
|
||||
if err := s.repo.MutateCloudAgent("user", root.ID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error {
|
||||
return cloudAgentSave(current, &state)
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Where("run_id = ?", root.ID).Delete(&model.CloudAgentEventRecord{}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
orphaned, err := s.repo.CloudAgent("user", root.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err = cloudAgentDecode(orphaned); err == nil || !strings.Contains(err.Error(), "journal is incomplete") {
|
||||
t.Fatalf("事件行全缺时仍被接受:%v", err)
|
||||
}
|
||||
}
|
||||
@@ -152,6 +152,11 @@ type cloudAgentRuntime struct {
|
||||
// HistoryIncludesCurrent 说明 TextHistory 是压缩瞬间的 canonical 快照:
|
||||
// 续轮只补快照之后的新插话与最终回复,不能重复已压缩的用户要求。
|
||||
HistoryIncludesCurrent bool `json:"historyIncludesCurrent,omitempty"`
|
||||
// EventSeqBase 是本次载入的事件窗口之前已入库的条数,不变量是
|
||||
// events[i].Seq == EventSeqBase + i + 1。事件全量在 cloud_agent_event_records,
|
||||
// 内存只保留最近一窗(repository.CloudAgentJournalWindow),因此它是"窗口在整条
|
||||
// 日志里的偏移",而不是累计条数。只在内存里有效,不进检查点(检查点里的事件为空)。
|
||||
EventSeqBase int `json:"-"`
|
||||
}
|
||||
|
||||
type cloudAgentTransientReference struct {
|
||||
@@ -196,7 +201,8 @@ func (s *Service) ensureCloudAgentExecution(task *model.Task, initial cloudAgent
|
||||
return s.repo.EnsureCloudAgent(run)
|
||||
}
|
||||
func (state *cloudAgentRuntime) event(id, kind string, payload map[string]any) {
|
||||
seq := len(state.Events) + 1
|
||||
// 序号接在"已载入窗口 + 窗口之前已入库条数"之后,与 EventSeqBase 的不变量一致。
|
||||
seq := state.EventSeqBase + len(state.Events) + 1
|
||||
state.Events = append(state.Events, CloudAgentEvent{EventID: fmt.Sprintf("%s:%d", id, seq), RunID: id, Seq: seq, Type: kind, Payload: payload, CreatedAt: time.Now()})
|
||||
}
|
||||
func cloudAgentDecode(run *model.CloudAgentExecution) (cloudAgentRuntime, error) {
|
||||
@@ -209,13 +215,25 @@ func cloudAgentDecode(run *model.CloudAgentExecution) (cloudAgentRuntime, error)
|
||||
}
|
||||
state.RuntimeRunID = run.ID
|
||||
if run.CheckpointVersion >= 2 {
|
||||
if len(run.Journal) != run.EventCount || len(run.Transcript) != run.MessageCount {
|
||||
if len(run.Transcript) != run.MessageCount {
|
||||
return state, errors.New("Agent execution transcript is incomplete")
|
||||
}
|
||||
// 事件只载入最近一窗(repository.CloudAgentJournalWindow),所以这里校验的是
|
||||
// "窗口与水位自洽"而不是全量条数:
|
||||
// - 窗口必须正好结束在水位上(最后一条 seq == EventCount),否则说明有行没读到;
|
||||
// - 窗口为空时水位只能是 0,否则同样说明行缺失(不能把"一行都没有"当成空日志);
|
||||
// - EventSeqBase = 水位 - 窗口条数,即窗口之前已入库的条数。
|
||||
if len(run.Journal) == 0 && run.EventCount != 0 {
|
||||
return state, errors.New("Agent execution journal is incomplete")
|
||||
}
|
||||
state.EventSeqBase = run.EventCount - len(run.Journal)
|
||||
if state.EventSeqBase < 0 {
|
||||
return state, errors.New("Agent execution journal watermark is invalid")
|
||||
}
|
||||
state.Events = make([]CloudAgentEvent, 0, len(run.Journal))
|
||||
state.Canonical.Messages, state.TextHistory = nil, nil
|
||||
for _, record := range run.Journal {
|
||||
expected := len(state.Events) + 1
|
||||
expected := state.EventSeqBase + len(state.Events) + 1
|
||||
if record.Sequence != expected {
|
||||
return state, errors.New("Agent event sequence is incomplete")
|
||||
}
|
||||
@@ -228,6 +246,9 @@ func cloudAgentDecode(run *model.CloudAgentExecution) (cloudAgentRuntime, error)
|
||||
}
|
||||
state.Events = append(state.Events, event)
|
||||
}
|
||||
if len(state.Events) > 0 && state.Events[len(state.Events)-1].Seq != run.EventCount {
|
||||
return state, errors.New("Agent execution journal is incomplete")
|
||||
}
|
||||
for _, record := range run.Transcript {
|
||||
switch record.Kind {
|
||||
case "canonical":
|
||||
@@ -372,8 +393,16 @@ func validateCloudAgentRuntime(run *model.CloudAgentExecution, state *cloudAgent
|
||||
if state.Decisions == nil || state.Events == nil {
|
||||
return errors.New("Agent runtime maps are missing")
|
||||
}
|
||||
if state.EventSeqBase < 0 {
|
||||
return errors.New("Agent runtime event watermark is invalid")
|
||||
}
|
||||
// 内存里只有"最近一窗 + 本次转移新产生的部分",超过健全上限说明窗口没有按
|
||||
// CloudAgentJournalWindow 载入(或序号基准算错),此时不能继续推进。
|
||||
if len(state.Events) > cloudAgentEventWindowSanityLimit {
|
||||
return errors.New("Agent runtime event window is too large")
|
||||
}
|
||||
for index, event := range state.Events {
|
||||
if event.RunID != run.ID || event.Seq != index+1 || event.EventID == "" || event.Type == "" || event.Payload == nil || event.CreatedAt.IsZero() {
|
||||
if event.RunID != run.ID || event.Seq != state.EventSeqBase+index+1 || event.EventID == "" || event.Type == "" || event.Payload == nil || event.CreatedAt.IsZero() {
|
||||
return errors.New("Agent runtime event history is invalid")
|
||||
}
|
||||
if err := validateCloudAgentID(event.EventID, "事件 ID", 240); err != nil {
|
||||
@@ -519,7 +548,7 @@ func cloudAgentSave(run *model.CloudAgentExecution, state *cloudAgentRuntime) er
|
||||
return fmt.Errorf("%w: %v", errCloudAgentCheckpoint, err)
|
||||
}
|
||||
for index, event := range state.Events {
|
||||
sequence := index + 1
|
||||
sequence := state.EventSeqBase + index + 1
|
||||
if event.Seq != sequence || event.RunID != run.ID || event.EventID != fmt.Sprintf("%s:%d", run.ID, sequence) {
|
||||
return fmt.Errorf("%w: Agent event sequence or identity is invalid", errCloudAgentCheckpoint)
|
||||
}
|
||||
@@ -541,6 +570,9 @@ func cloudAgentSave(run *model.CloudAgentExecution, state *cloudAgentRuntime) er
|
||||
if run.Title == "" {
|
||||
run.Title = truncateRunes(state.Request.Prompt, 80)
|
||||
}
|
||||
// 事件按"窗口 + 水位"落库:run.Journal 只是内存窗口的映射(窗口之外的历史仍在
|
||||
// 事件表里),写入侧(repository.MutateCloudAgent)只追加 seq > 旧水位的行,
|
||||
// 因此窗口左移不会丢历史。
|
||||
run.Journal = make([]model.CloudAgentEventRecord, 0, len(state.Events))
|
||||
for _, event := range state.Events {
|
||||
body, err := json.Marshal(event)
|
||||
@@ -564,11 +596,13 @@ func cloudAgentSave(run *model.CloudAgentExecution, state *cloudAgentRuntime) er
|
||||
}
|
||||
run.Transcript = append(run.Transcript, model.CloudAgentMessageRecord{RunID: run.ID, UserID: run.UserID, Kind: "history", Sequence: index + 1, MessageJSON: string(body)})
|
||||
}
|
||||
run.CheckpointVersion, run.EventCount, run.MessageCount = 2, len(run.Journal), len(run.Transcript)
|
||||
// 水位 = 窗口之前已入库的条数 + 本次载入/新增的窗口条数,必须与事件表里的
|
||||
// 最大 seq 一致,否则下一次载入的窗口与水位会对不上。
|
||||
run.CheckpointVersion, run.EventCount, run.MessageCount = 2, state.EventSeqBase+len(state.Events), len(run.Transcript)
|
||||
run.StateJSON = string(raw)
|
||||
return nil
|
||||
}
|
||||
func (s *Service) cloudAgentExecutionOutput(task *model.Task, initial cloudAgentState) (*CloudAgentRun, error) {
|
||||
func (s *Service) cloudAgentExecutionOutput(task *model.Task, initial cloudAgentState, options ...CloudAgentRunViewOptions) (*CloudAgentRun, error) {
|
||||
run, err := s.repo.CloudAgent(task.UserID, task.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -584,7 +618,27 @@ func (s *Service) cloudAgentExecutionOutput(task *model.Task, initial cloudAgent
|
||||
out.Status = run.Status
|
||||
out.Revision, out.CleanupPending, out.FailureMessage = run.Revision, run.CleanupPending, run.FailureMessage
|
||||
out.UpdatedAt = run.UpdatedAt
|
||||
out.Events = state.Events
|
||||
view := CloudAgentRunViewOptions{}
|
||||
if len(options) > 0 {
|
||||
view = options[0]
|
||||
}
|
||||
// 事件已全量落库,运行详情只返回一页:默认是尾部窗口,sinceSeq 只取增量。
|
||||
// 四个位置字段与 events 一起返回,客户端据此判断"是否还有更早的记录"。
|
||||
out.Events = s.cloudAgentRunEventsForView(task.UserID, run, &state, view.SinceSeq, view.EventLimit)
|
||||
// EventSeqBase 必须与真正返回的这一页对齐(eventLimit 把它收窄时也一样),
|
||||
// 不变量 events[i].seq == eventSeqBase + i + 1 才成立。页为空时退回窗口水位:
|
||||
// 那时没有"首条事件",但仍然要能说明窗口在整条日志里的位置。
|
||||
out.EventSeqBase = state.EventSeqBase
|
||||
if len(out.Events) > 0 {
|
||||
out.EventSeqBase = out.Events[0].Seq - 1
|
||||
}
|
||||
out.EventCount = s.cloudAgentRunEventCount(task.UserID, run, &state)
|
||||
if len(out.Events) > 0 {
|
||||
out.LatestSeq = out.Events[len(out.Events)-1].Seq
|
||||
}
|
||||
if view.SinceSeq == 0 && len(out.Events) < out.EventCount {
|
||||
out.EventsTruncated = true
|
||||
}
|
||||
out.Approval = state.Approval
|
||||
if cloudAgentRunTerminal(run.Status) {
|
||||
out.Approval = nil
|
||||
|
||||
@@ -5,7 +5,9 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
@@ -195,7 +197,14 @@ func RegisterAgentRoutes(r *gin.RouterGroup, svc *service.Service) {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
run, err := svc.CloudAgentRun(user.ID, c.Param("id"))
|
||||
// 运行详情只返回一页事件,分页参数越界或非法一律拒绝:静默夹取会让客户端
|
||||
// 以为拿到的是它请求的那一页,从而把缺口当成"没有更多记录"。
|
||||
options, err := agentRunViewOptions(c)
|
||||
if err != nil {
|
||||
fail(c, http.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
run, err := svc.CloudAgentRun(user.ID, c.Param("id"), options)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
@@ -297,16 +306,19 @@ func RegisterAgentRoutes(r *gin.RouterGroup, svc *service.Service) {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
run, err := svc.CloudAgentRun(user.ID, c.Param("id"))
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
// 游标就是事件 seq:先解析它,才能只把 after 之后的增量交给视图,
|
||||
// 断线重连不必重新载入前面已经发过的事件。
|
||||
after, err := taskTextEventCursor(c)
|
||||
if err != nil {
|
||||
fail(c, 400, err)
|
||||
return
|
||||
}
|
||||
options := service.CloudAgentRunViewOptions{SinceSeq: int(after)}
|
||||
run, err := svc.CloudAgentRun(user.ID, c.Param("id"), options)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
c.Header("Content-Type", "text/event-stream")
|
||||
c.Header("Cache-Control", "no-cache")
|
||||
c.Header("X-Accel-Buffering", "no")
|
||||
@@ -345,7 +357,7 @@ func RegisterAgentRoutes(r *gin.RouterGroup, svc *service.Service) {
|
||||
case <-c.Request.Context().Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
run, err = svc.CloudAgentRunIfChanged(user.ID, c.Param("id"), revision)
|
||||
run, err = svc.CloudAgentRunIfChanged(user.ID, c.Param("id"), revision, options)
|
||||
}
|
||||
}
|
||||
})
|
||||
@@ -362,3 +374,31 @@ func writeAgentSSE(c *gin.Context, event string, id int64, value any) {
|
||||
_, _ = fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, data)
|
||||
c.Writer.Flush()
|
||||
}
|
||||
|
||||
// agentRunEventQueryLimit 是 eventLimit 允许的最大值,与 app 侧的增量上限一致:
|
||||
// 请求更大的页没有意义(单次响应体积),直接拒绝比静默截断更容易被发现。
|
||||
const agentRunEventQueryLimit = 500
|
||||
|
||||
func agentRunViewOptions(c *gin.Context) (service.CloudAgentRunViewOptions, error) {
|
||||
options := service.CloudAgentRunViewOptions{}
|
||||
for _, item := range []struct {
|
||||
name string
|
||||
value *int
|
||||
min int
|
||||
max int
|
||||
}{
|
||||
{name: "sinceSeq", value: &options.SinceSeq, min: 0, max: math.MaxInt32},
|
||||
{name: "eventLimit", value: &options.EventLimit, min: 1, max: agentRunEventQueryLimit},
|
||||
} {
|
||||
raw := strings.TrimSpace(c.Query(item.name))
|
||||
if raw == "" {
|
||||
continue
|
||||
}
|
||||
parsed, err := strconv.Atoi(raw)
|
||||
if err != nil || parsed < item.min || parsed > item.max {
|
||||
return options, fmt.Errorf("%s 必须是 %d–%d 之间的整数", item.name, item.min, item.max)
|
||||
}
|
||||
*item.value = parsed
|
||||
}
|
||||
return options, nil
|
||||
}
|
||||
|
||||
63
backend/internal/handler/agent_run_pagination_test.go
Normal file
63
backend/internal/handler/agent_run_pagination_test.go
Normal file
@@ -0,0 +1,63 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// 运行详情分页参数(?sinceSeq= / ?eventLimit=)。合法值按原样进入视图选项,
|
||||
// 非法或越界一律拒绝:静默夹取会让客户端以为拿到的是它请求的那一页,
|
||||
// 从而把缺口当成"没有更多记录"。
|
||||
func TestAgentRunViewOptionsParsesPaginationQuery(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
for _, testCase := range []struct {
|
||||
query string
|
||||
wantSinceSeq, wantEventLimit int
|
||||
}{
|
||||
{query: "", wantSinceSeq: 0, wantEventLimit: 0},
|
||||
{query: "?sinceSeq=0", wantSinceSeq: 0, wantEventLimit: 0},
|
||||
{query: "?sinceSeq=12", wantSinceSeq: 12, wantEventLimit: 0},
|
||||
{query: "?eventLimit=1", wantSinceSeq: 0, wantEventLimit: 1},
|
||||
{query: "?eventLimit=500", wantSinceSeq: 0, wantEventLimit: 500},
|
||||
{query: "?sinceSeq=12&eventLimit=50", wantSinceSeq: 12, wantEventLimit: 50},
|
||||
{query: "?sinceSeq=%203%20&eventLimit=%207%20", wantSinceSeq: 3, wantEventLimit: 7},
|
||||
} {
|
||||
t.Run(testCase.query, func(t *testing.T) {
|
||||
request := httptest.NewRequest("GET", "/api/agent/runs/run-1"+testCase.query, nil)
|
||||
context, _ := gin.CreateTestContext(httptest.NewRecorder())
|
||||
context.Request = request
|
||||
options, err := agentRunViewOptions(context)
|
||||
if err != nil {
|
||||
t.Fatalf("%s 被拒绝: %v", testCase.query, err)
|
||||
}
|
||||
if options.SinceSeq != testCase.wantSinceSeq || options.EventLimit != testCase.wantEventLimit {
|
||||
t.Fatalf("%s 解析为 %+v,期望 sinceSeq=%d eventLimit=%d", testCase.query, options, testCase.wantSinceSeq, testCase.wantEventLimit)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentRunViewOptionsRejectsInvalidPaginationQuery(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
for _, query := range []string{
|
||||
"?sinceSeq=abc",
|
||||
"?sinceSeq=-1",
|
||||
"?sinceSeq=1.5",
|
||||
"?sinceSeq=99999999999999999999",
|
||||
"?eventLimit=0",
|
||||
"?eventLimit=501",
|
||||
"?eventLimit=-5",
|
||||
"?eventLimit=abc",
|
||||
} {
|
||||
t.Run(query, func(t *testing.T) {
|
||||
request := httptest.NewRequest("GET", "/api/agent/runs/run-1"+query, nil)
|
||||
context, _ := gin.CreateTestContext(httptest.NewRecorder())
|
||||
context.Request = request
|
||||
if options, err := agentRunViewOptions(context); err == nil {
|
||||
t.Fatalf("%s 未被拒绝,解析为 %+v", query, options)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -679,9 +679,23 @@ paths:
|
||||
get:
|
||||
tags: [agent]
|
||||
summary: 读取当前用户的 Agent 轮次
|
||||
description: data.run 只返回一页事件(默认最新 40 条)并带上 eventSeqBase、eventCount、latestSeq、eventsTruncated:eventSeqBase 是本次首条事件之前已入库的条数(不变量 events[i].seq == eventSeqBase + i + 1),latestSeq 可直接作为下次 sinceSeq,eventsTruncated 表示还有更早的记录未随本次返回。读取走 (run_id, sequence) 的 keyset 分页,不使用 offset;升级前把事件放在 state_json 里的轮次在事件表中没有行,读取回退到内存窗口。
|
||||
parameters:
|
||||
- in: query
|
||||
name: sinceSeq
|
||||
required: false
|
||||
description: 只返回该事件序号之后的事件(非负整数,单次最多 500 条);追平或越界时返回空 events
|
||||
schema: { type: integer, minimum: 0 }
|
||||
- in: query
|
||||
name: eventLimit
|
||||
required: false
|
||||
description: 单次返回的事件条数上限,1–500,默认为尾部窗口大小
|
||||
schema: { type: integer, minimum: 1, maximum: 500 }
|
||||
responses:
|
||||
"200":
|
||||
description: 信封 data.run
|
||||
description: 信封 data.run(含 events 与 eventSeqBase / eventCount / latestSeq / eventsTruncated)
|
||||
"400":
|
||||
description: sinceSeq 或 eventLimit 不是整数或超出范围
|
||||
"404":
|
||||
description: 轮次不存在或不属于当前用户
|
||||
/agent/runs/{id}/messages:
|
||||
|
||||
@@ -38,19 +38,72 @@ func (r *Repository) CloudAgent(userID, id string) (*model.CloudAgentExecution,
|
||||
return &run, err
|
||||
}
|
||||
|
||||
// CloudAgentJournalWindow 是载入执行状态时读取的事件窗口条数。
|
||||
//
|
||||
// 事件全量在 cloud_agent_event_records(append-only,主键 run_id + sequence),
|
||||
// 内存只保留最近这一窗:摘要、卡死判定与个人记忆提取只看最近事件,运行详情与
|
||||
// SSE 需要更早的记录时按 seq 分页读表。若在这里全量载入,长跑的 run 每读一次
|
||||
// 运行详情(以及调度器每轮扫描的每个活动 run)都要把整份 journal 拉进内存。
|
||||
const CloudAgentJournalWindow = 40
|
||||
|
||||
func (r *Repository) hydrateCloudAgent(run *model.CloudAgentExecution) error {
|
||||
if run.CheckpointVersion < 2 {
|
||||
return nil
|
||||
}
|
||||
if err := r.db.Where("run_id = ? AND user_id = ?", run.ID, run.UserID).Order("sequence").Find(&run.Journal).Error; err != nil {
|
||||
// 只取最近一窗,再翻回升序:与 EventCount 的尾部对齐(不变量见 app 侧解码)。
|
||||
window := []model.CloudAgentEventRecord{}
|
||||
if err := r.db.Where("run_id = ? AND user_id = ?", run.ID, run.UserID).
|
||||
Order("sequence DESC").Limit(CloudAgentJournalWindow).Find(&window).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
run.Journal = make([]model.CloudAgentEventRecord, 0, len(window))
|
||||
for index := len(window) - 1; index >= 0; index-- {
|
||||
run.Journal = append(run.Journal, window[index])
|
||||
}
|
||||
if err := r.db.Where("run_id = ? AND user_id = ?", run.ID, run.UserID).Order("kind, sequence").Find(&run.Transcript).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// CloudAgentEventRecords 取 seq > afterSeq 的事件(升序),keyset 分页走
|
||||
// (run_id, sequence) 主键,用于增量拉取与 SSE 断线重连。
|
||||
func (r *Repository) CloudAgentEventRecords(userID, runID string, afterSeq, limit int) ([]model.CloudAgentEventRecord, error) {
|
||||
if limit <= 0 {
|
||||
limit = CloudAgentJournalWindow
|
||||
}
|
||||
records := []model.CloudAgentEventRecord{}
|
||||
err := r.db.Where("run_id = ? AND user_id = ? AND sequence > ?", runID, userID, afterSeq).
|
||||
Order("sequence ASC").Limit(limit).Find(&records).Error
|
||||
return records, err
|
||||
}
|
||||
|
||||
// CloudAgentEventRecordsBefore 取 seq < beforeSeq 的**最近** limit 条(升序返回),
|
||||
// 供"请求的页比内存窗口更深、要往前补"的读路径使用。
|
||||
func (r *Repository) CloudAgentEventRecordsBefore(userID, runID string, beforeSeq, limit int) ([]model.CloudAgentEventRecord, error) {
|
||||
if limit <= 0 {
|
||||
limit = CloudAgentJournalWindow
|
||||
}
|
||||
records := []model.CloudAgentEventRecord{}
|
||||
err := r.db.Where("run_id = ? AND user_id = ? AND sequence < ?", runID, userID, beforeSeq).
|
||||
Order("sequence DESC").Limit(limit).Find(&records).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for left, right := 0, len(records)-1; left < right; left, right = left+1, right-1 {
|
||||
records[left], records[right] = records[right], records[left]
|
||||
}
|
||||
return records, nil
|
||||
}
|
||||
|
||||
// CloudAgentEventRecordCount 是该运行已入库的事件条数;只做计数,不载入正文。
|
||||
func (r *Repository) CloudAgentEventRecordCount(userID, runID string) (int64, error) {
|
||||
var count int64
|
||||
err := r.db.Model(&model.CloudAgentEventRecord{}).
|
||||
Where("run_id = ? AND user_id = ?", runID, userID).Count(&count).Error
|
||||
return count, err
|
||||
}
|
||||
|
||||
func (r *Repository) CloudAgentForActiveTask(userID, taskID string) (*model.CloudAgentExecution, error) {
|
||||
var run model.CloudAgentExecution
|
||||
err := r.db.Where("user_id = ? AND active_task_id = ? AND status IN ?", userID, taskID, []string{"running", "queued"}).First(&run).Error
|
||||
@@ -144,14 +197,34 @@ func (r *Repository) MutateCloudAgent(userID, id string, revision int64, fn func
|
||||
if err = fn(run, New(tx)); err != nil {
|
||||
return err
|
||||
}
|
||||
if run.EventCount < previousEvents || len(run.Journal) != run.EventCount {
|
||||
if run.EventCount < previousEvents {
|
||||
return fmt.Errorf("cloud Agent journal cannot be truncated")
|
||||
}
|
||||
// journal 是 append-only:本次转移之前已经载入的事件不许改写。
|
||||
// 内存里只有最近一窗,窗口右移后更早的行会落在新窗口之外——那些行不在
|
||||
// run.Journal 里,也不会被下面的 Create 循环碰到(只追加 seq > 水位的新行),
|
||||
// 因此只要求它们确实比新窗口更早,而不是要求它们仍在窗口内。
|
||||
current := make(map[int]string, len(run.Journal))
|
||||
for _, event := range run.Journal {
|
||||
current[event.Sequence] = event.EventJSON
|
||||
}
|
||||
for sequence, body := range previousEventBodies {
|
||||
if sequence > len(run.Journal) || run.Journal[sequence-1].Sequence != sequence || !sameJSONDocument(run.Journal[sequence-1].EventJSON, body) {
|
||||
updated, present := current[sequence]
|
||||
if !present {
|
||||
if len(run.Journal) == 0 || sequence >= run.Journal[0].Sequence {
|
||||
return fmt.Errorf("cloud Agent journal is append-only")
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !sameJSONDocument(updated, body) {
|
||||
return fmt.Errorf("cloud Agent journal is append-only")
|
||||
}
|
||||
}
|
||||
// 水位与窗口必须自洽:窗口最后一条就是要落库的最后一条,否则说明写入端
|
||||
// 算错了窗口起点,或行被删掉过。
|
||||
if len(run.Journal) > 0 && run.Journal[len(run.Journal)-1].Sequence != run.EventCount {
|
||||
return fmt.Errorf("cloud Agent journal watermark is inconsistent")
|
||||
}
|
||||
if err = tx.Omit("Journal", "Transcript").Save(run).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ type (
|
||||
PaymentOrderQuery = app.PaymentOrderQuery
|
||||
PaymentReconciliationQuery = app.PaymentReconciliationQuery
|
||||
CloudAgentRequest = app.CloudAgentRequest
|
||||
CloudAgentRunViewOptions = app.CloudAgentRunViewOptions
|
||||
CloudAgentMediaSettings = app.CloudAgentMediaSettings
|
||||
CloudAgentCapabilitySet = app.CloudAgentCapabilitySet
|
||||
AgentProfileRequest = app.AgentProfileRequest
|
||||
|
||||
@@ -224,8 +224,10 @@ Token 报价的 `estimated` 为 `true`,`quantity` 是预留 Token 数。参考
|
||||
- `canvas_get_state` 首次调用传 `{}`,仅接受可选的 `offset`、`storyboardOffset`(非负整数)和 `nodeIds`(最多 8 个字符串组成的数组);当前画布由运行绑定,不接受 `canvasId`、`limit` 或 `page`。读取工具分别反馈非对象 / 多对象、JSON 格式错误、未知字段和字段类型错误。可纠正的工具参数失败在 `tool_failed.result` 及模型工具结果中返回 `reason: invalid_tool_arguments`、本轮已声明的 `parameters` 和 `guidance`;画布读取额外给出 `exampleArguments: {}`。失败调用仍如实记录,由模型修正参数后继续,不静默丢弃字段或自动重放写操作;资源归属、审批、快照和收费校验仍按原规则执行。
|
||||
- `GET /agent/runs/:id` 返回运行信息;`POST /agent/runs/:id/cancel` 取消当前运行及活动子任务;`POST /agent/runs/:id/approvals/:approvalId/decision` 接受 `{"decision":"approve"|"reject","reason":"..."}`,审批决定幂等且服务端校验归属。
|
||||
- `GET /agent/skills/usage` 需要登录,只聚合当前用户最近最多 200 次 Agent 运行的日志;返回技能启用次数、检索命中、成功/失败读取、总纲/卡片读取和最多 5 条最常读卡片路径,不返回技能正文。损坏的日志会返回错误而非静默给出不完整统计。
|
||||
- 运行详情的事件**分页**:事件全量在 `cloud_agent_event_records`(append-only,主键 `run_id` + `sequence`),运行详情默认只返回**尾部一窗**(最新 40 条),并同时给出四个位置字段:`eventSeqBase`(本次返回的首条事件之前已入库的条数,不变量是 `events[i].seq == eventSeqBase + i + 1`)、`eventCount`(该轮次累计事件条数)、`latestSeq`(本次返回的最后一条序号,可直接作为下次 `sinceSeq`)、`eventsTruncated`(还有更早的记录没随本次返回)。不返回事件时 `eventsTruncated` 为 false,`eventSeqBase` 退回内存窗口在整条日志里的偏移。
|
||||
- 运行详情查询参数:`?sinceSeq=<seq>` 只返回该序号**之后**的事件(增量 / 断线重连,单次最多 500 条,按 seq 升序);`?eventLimit=<n>` 覆盖默认页大小,取值 1–500,大于窗口时由服务端从事件表往前补页。两个参数都必须是整数,越界或非法返回 400;游标追平或越界时返回空 `events` 而不是报错。读取一律走 `(run_id, sequence)` 的 keyset 分页(`sequence > ? ORDER BY sequence LIMIT ?`),不用 offset,也不会为这些字段载入整条日志。升级前把事件放在 `state_json` 里的旧轮次在事件表中没有行,读取回退到内存窗口,行为与升级前一致。
|
||||
- 首次通过取消接口取消运行时记录一次 `run_cancelled` 事件,`source: user_request` 表示来自用户取消 API,并携带当时的 `activeTaskId` / `mediaTaskId`;此事件表示取消请求已接收,关联任务最终状态以任务查询及收尾状态为准。运行记录损坏或超限时优先保证取消和后台收尾,事件无法保存会记录服务端告警;旧导出日志不能据此追溯取消发起来源。
|
||||
- `GET /agent/runs/:id/events` 返回 `agent_event` 和首次 / 状态版本变化时的紧凑 `run_snapshot` SSE;无变化时只发送 heartbeat。事件 `Seq` 在运行内递增;支持 `?after=` / `Last-Event-ID`,游标不是运行 ID。`cleanupPending: true` 时即使运行已进入终态也会继续推送,直到后台收尾完成。前端应以快照恢复状态,代理只对这一精确路径关闭缓冲。
|
||||
- `GET /agent/runs/:id/events` 返回 `agent_event` 和首次 / 状态版本变化时的紧凑 `run_snapshot` SSE;无变化时只发送 heartbeat。事件 `Seq` 在运行内递增;支持 `?after=` / `Last-Event-ID`,游标不是运行 ID,服务端把它作为运行详情的 `sinceSeq`,因此断线重连只读该游标之后的增量。`cleanupPending: true` 时即使运行已进入终态也会继续推送,直到后台收尾完成。前端应以快照恢复状态,代理只对这一精确路径关闭缓冲。
|
||||
- `generation_task_created` 携带 `canvasId`、`nodeId`、`taskId`、`title`、`mode`、`referenceNodeIds`。草稿、提交、画布工具写入,以及媒体成功 / 失败 / 取消的真实节点变更,均在画布写入与运行检查点的同一事务持久化 `canvas_updated`。节点已删除或换任务时不伪造更新;`run_failed` 本身不代表节点媒体任务失败。
|
||||
- 未注册 Agent 能力的节点不阻断整张画布读取:摘要和 `canvas_get_state` 仅返回基础信息,并标记 `agentSupported: false` 与原因;不暴露任意 metadata,不授予该类型的创建、修改、连线或媒体引用能力。
|
||||
- `approval.preview` 和 `canvas_updated.actions` 是服务端根据同一解析、快照与 Capability 规划流程生成的解释信息,不是授权凭证。审批卡片明确列出新增/修改节点、节点类型、修改字段、名称变化和连线两端;执行事件保存同样的可读信息,供历史消息显示和点击定位。实际写入仍重新校验用户归属、`snapshotHash`、Capability、节点锁、预算和审批 `callHash`。
|
||||
|
||||
Reference in New Issue
Block a user