diff --git a/backend/internal/app/cloud_agent.go b/backend/internal/app/cloud_agent.go index 4f8536b40..a0fa597a7 100644 --- a/backend/internal/app/cloud_agent.go +++ b/backend/internal/app/cloud_agent.go @@ -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 } diff --git a/backend/internal/app/cloud_agent_continuation.go b/backend/internal/app/cloud_agent_continuation.go index 0f6d5057c..7a499998b 100644 --- a/backend/internal/app/cloud_agent_continuation.go +++ b/backend/internal/app/cloud_agent_continuation.go @@ -17,6 +17,10 @@ const ( // cloudAgentContinuationFrameBytes 是交接帧的硬上限：超了就按"最近的改动优先"裁剪， // 并把裁掉的组数记进 omittedCanvasChanges，绝不把整份历史顶过硬闸。 cloudAgentContinuationFrameBytes = 8 << 10 + // cloudAgentContinuationEventLimit 是续轮读取上一轮事件的上限。运行详情默认只返回尾部 + // 一窗，而收束要覆盖上一轮**全部**真实改动（画布改动、已提交任务、助手原话）， + // 被窗口截断就会表现为"新一轮忘了上一轮做过什么"。 + cloudAgentContinuationEventLimit = 1000 ) type cloudAgentContinuationChange struct { diff --git a/backend/internal/app/cloud_agent_event_store.go b/backend/internal/app/cloud_agent_event_store.go new file mode 100644 index 000000000..2eadc255a --- /dev/null +++ b/backend/internal/app/cloud_agent_event_store.go @@ -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 +} diff --git a/backend/internal/app/cloud_agent_event_store_test.go b/backend/internal/app/cloud_agent_event_store_test.go new file mode 100644 index 000000000..a5b1875d4 --- /dev/null +++ b/backend/internal/app/cloud_agent_event_store_test.go @@ -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) + } +} diff --git a/backend/internal/app/cloud_agent_runtime.go b/backend/internal/app/cloud_agent_runtime.go index 463a25fdf..403698ab8 100644 --- a/backend/internal/app/cloud_agent_runtime.go +++ b/backend/internal/app/cloud_agent_runtime.go @@ -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 diff --git a/backend/internal/handler/agent.go b/backend/internal/handler/agent.go index 5cdab45d8..927b4f15b 100644 --- a/backend/internal/handler/agent.go +++ b/backend/internal/handler/agent.go @@ -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 +} diff --git a/backend/internal/handler/agent_run_pagination_test.go b/backend/internal/handler/agent_run_pagination_test.go new file mode 100644 index 000000000..e2400db2c --- /dev/null +++ b/backend/internal/handler/agent_run_pagination_test.go @@ -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) + } + }) + } +} diff --git a/backend/internal/handler/openapi.yaml b/backend/internal/handler/openapi.yaml index 022b4aa4f..4dab2d3c1 100644 --- a/backend/internal/handler/openapi.yaml +++ b/backend/internal/handler/openapi.yaml @@ -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: diff --git a/backend/internal/repository/cloud_agent.go b/backend/internal/repository/cloud_agent.go index 15f416ab7..188a1491e 100644 --- a/backend/internal/repository/cloud_agent.go +++ b/backend/internal/repository/cloud_agent.go @@ -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 } diff --git a/backend/internal/service/aliases_types.go b/backend/internal/service/aliases_types.go index bb9239964..c1ccad367 100644 --- a/backend/internal/service/aliases_types.go +++ b/backend/internal/service/aliases_types.go @@ -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 diff --git a/docs/content/docs/backend/http-api.mdx b/docs/content/docs/backend/http-api.mdx index fbb99e458..fd19d17c3 100644 --- a/docs/content/docs/backend/http-api.mdx +++ b/docs/content/docs/backend/http-api.mdx @@ -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=` 只返回该序号**之后**的事件（增量 / 断线重连，单次最多 500 条，按 seq 升序）；`?eventLimit=` 覆盖默认页大小，取值 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`。