diff --git a/Docs/api-client.md b/Docs/api-client.md index c4af883..e89ba3e 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -28,7 +28,7 @@ | 3 | `/api/v1/chat` | POST | JWT | ✅ | `{text,image?,deviceId?,mode?,stream?}` → `{data:{reply}}`;`stream:true` 时经 WS 触发流式。`image` 为 base64 data: URI 或 http(s) URL,入视觉路由(design/06)——**真视觉需配 `LLM_VISION_MODEL`/凭据,无凭据时降级为主模型直接看图或视觉·OCR 工具文本描述**;`deviceId` 为**客户端生成的具体端标识**(design/16 §4.5,如 `web:chrome:a1b2c3`,可选),落库到消息、便于主人知道从哪个端发来;`mode` 为**行为模式**(design/01 §3 / design/16 §5.1,如 `chat`/`code`/`home`/`docs`/…,可选),随该轮次落库——IM 恒 `chat`、缺省回落 `chat`,供前端按模式渲染气泡(仅 `chat` 才解析 `

` 拆气泡)。**会话编排(design/15)**:非流式且该会话「正忙」时**不阻塞**,返回 `202` `{data:{queued:true,queueDepth}}`(稍后按上下文续接);待处理队列满时返回 `409` `{error:{code:"chat_busy"}}` | | 4 | `/api/v1/sessions` | GET | JWT | ✅ | `{data:[session...]}`。**会话持久化(design/16)**:注入 HistoryStore 后返回持久化会话元数据 `{data:[{id,type("main"|"sub"),title?,createdAt,lastMessage?}]}`(主会话 `id="web"` 恒在首位,次级 `web:`;`lastMessage` 为**最后一条消息预览** `{content,role(user|assistant),createdAt}`,content 折叠空白截前 60 rune,空会话省略);未注入时退化回渠道会话列表 | | 5 | `/api/v1/sessions` | POST | JWT(owner/admin) | ✅ | **新建次级会话** → `201 {data:{sessionId}}`(`sessionId="web:"`);未注入 HistoryStore 回 `501 not_implemented` | -| 6 | `/api/v1/sessions/{id}/messages` | GET | JWT(owner/admin) | ✅ | **热分页拉历史**:`?limit=&before=&platform=` 默认 `limit=50`(上限 200)、`before`=上一页最旧 `createdAt`(毫秒),按时间倒序 `{data:[{id,sessionId,role,content,mode?,mood?,image?,thinking?,platform?,deviceId?,senderName?,createdAt}]}`;`platform` 为**可选**来源平台过滤(design/16 §4.4)——Web 端不传(=全量共享主会话),IM 适配器传**本平台**(如 `telegram`)只拉本渠道消息;`deviceId` 为**客户端具体端标识**(design/16 §4.5,如 `web:chrome:a1b2c3` / `weixin-ilink:conf0:wx_uid`);`senderName` 为**第三方发送者昵称**(design/16 §4.6,群聊/陌生人私聊消息带,主人私聊为空,`omitempty`);非法 `limit`/`before` 回 `400`;未注入 HistoryStore 回 `501` | +| 6 | `/api/v1/sessions/{id}/messages` | GET | JWT(owner/admin) | ✅ | **热分页拉历史**:`?limit=&before=&platform=&after=`(A5)默认 `limit=50`(上限 200)。**`before`**=上一页最旧 `createdAt`(毫秒),按时间**倒序**;**`after`**=<`createdAt` 毫秒>,按时间**升序**返回 `created_at > after` 的消息(**正向增量**,客户端把上次看到的 `createdAt` 当游标补断线期间漏的,解决重连只回放 50 条的缺口);**`after` 与 `before` 互斥**,同传回 `400`;负数/非数字回 `400`。`platform` 为**可选**来源平台过滤(design/16 §4.4)——Web 端不传(=全量共享主会话),IM 适配器传**本平台**(如 `telegram`)只拉本渠道消息;`deviceId` 为**客户端具体端标识**;`senderName` 为**第三方发送者昵称**;非法 `limit`/`before`/`after` 回 `400`;未注入 HistoryStore 回 `501` | | 7 | `/api/v1/sessions/{id}` | DELETE | JWT(owner/admin) | ✅ | **删除次级会话及其消息** → `{data:{deleted:}}`;主会话不可删回 `403 forbidden`;未注入 HistoryStore 回 `501` | | 8 | `/api/v1/sessions/{id}` | PATCH | JWT(owner/admin) | ✅ | **会话改名**(A4):body `{title}`(trim、≤80 rune,空/超长回 `400`)→ `{data:{id,title}}`;不存在回 `404`;主会话(`id="web"`)可改名但不可删;未注入 HistoryStore 回 `501`。另:会话**首条用户消息到达时自动命名** title(取前 20 rune 折叠空白;纯空白回落 `主会话`/`次级会话`),列表不再空名 | | 9 | `/api/v1/personas` | GET | JWT | ✅ | `{data:[persona...]}` | @@ -215,6 +215,7 @@ - 语音:上行 `voice_input`(base64 音频或已转写 text),下行 `voice_output`(合成音频 base64 + text)。 ## 5. 变更记录 +- `v0.26`(**A5 消息正向增量**):`GET /api/v1/sessions/{id}/messages` 新增 **`after`** 查询参数(``,升序返回 `created_at > after` 的消息,默认 `limit=50`/上限 200)——解决「WS 重连只回放主会话最近 50 条、断线超 50 条永久丢」的缺口;`before` 现行为不变;**`after` 与 `before` 互斥**(同传 `400`);负数/非数字 `400`。`HistoryStore` 新增 `ListAfter`(走 `idx_chat_msgs_session_created`,`rowid` 兜底同毫秒并列)。**游标契约**:客户端用「本页最后一条 `createdAt`」作下一页 `after`;同毫秒消息 > limit 时纯 `createdAt` 游标会跨页重读(超罕见,客户端按 `id` 去重兜底,与现有绕行一致)。 - `v0.25`(**B2 refresh token**):`POST /api/v1/auth/refresh`(**新增**,`{refreshToken}` → 新 access + 新 refresh,默认单次使用旋转,并发重放 fail-closed `invalid_grant`)+ `POST /api/v1/auth/logout`(**新增**,撤单个或全设备)。`POST /api/v1/auth/login` 在 `{token,expiresAt,role}` 之外**新增 `refreshToken`**(不透明 256-bit 随机串,**仅 SHA-256 哈希落库**,`AUTH_REFRESH_TTL` 默认 `720h`,`AUTH_REFRESH_ROTATE_DISABLED` 关旋转)。新表 `refresh_tokens`(迁移 0008,id/user_id/role/token_hash UNIQUE/expires_at/revoked_at;含过期/轮换原子事务);`rotate-jwt` 成功即删除全部 refresh(返回 `refreshTokensRevoked`),`revoke-keys` 不动 refresh。**密码/secret 记录见 `internal/auth`;refresh 全文绝不落日志/审计。** - `v0.24`(**B1 图片上传端点**):`POST /api/v1/upload`(**新增**,multipart `file`,owner/admin+审计 → `201 {data:{url,size,mime,name}}`)+ `GET /api/v1/files/{name}`(**新增**,同端口静态服务)。安全:`MaxBytesReader`(5MB)、`DetectContentType` 嗅探(不信客户端 Content-Type)、白名单 raster image(拒 SVG/HTML)、`crypto/rand` 128-bit 随机名(丢弃原始文件名)、非可执行上传目录(0700)、`nosniff`;路径穿越全形态拒绝(含编码 `%2f`/`%2e%2e`)。错误码 `400/413/415`(`internal/errors` 增 `ErrPayloadTooLarge`/`ErrUnsupportedMediaType`)。**设计取舍**:GET 不鉴权(能力 URL,`chat.image`/`` 无法带 Bearer),POST 仍 owner/admin。`url` 为绝对地址(`X-Forwarded-Proto`/TLS + `r.Host`,因 `ParseImageRef` 只认 `http(s)://`)。`Deps.DataDir` 由 wire 注入(`resolveDataDir`)。 - `v0.23`(**C2 语音 silent 降级信号**):三处统一用 **`provider`** 只在「结果为空」时出现(非空/成功时才省略)——`POST /api/v1/audio/transcribe` 空文本响应体带 `{text:"", provider:"silent"}`;WS `voice_output` 空音频 payload 带 `provider`;`POST /api/v1/audio/speech` 空音频仍 `500 tts_empty` 并加 `X-Voice-Provider: silent` 响应头(CORS `Access-Control-Expose-Headers` 已放开)。`voice.Chain/Service` 新增 `STTWithProvider/TTSWithProvider`(返回真实档位 cloud/local/edge-tts/espeak/silent),网关用可选接口 `voiceProviderReporter` 透传,自定义实现回退 `silent`。**含 `X-Voice-Provider`(`apperr` 包装、`providerSignal`);`DASHSCOPE_API_KEY` 无/降级时信号为 `silent`。**(`go test -race` 在本机受内核 TSAN VMA 限制失败,非本代码问题。) diff --git a/Docs/api-reference.md b/Docs/api-reference.md index 16993f0..e829b4d 100644 --- a/Docs/api-reference.md +++ b/Docs/api-reference.md @@ -297,8 +297,9 @@ curl -s http://:8080/api/v1/personas -H "Authorization: Bearer $TOKEN" -H |------|------|:---:|------|------| | `limit` | int | 否 | 50 | 页大小,需 `>=1`,钳制上限 200(`maxPageLimit`);非法回 400 | | `before` | int64 | 否 | 0 | 游标:上一页最旧消息的 `createdAt`(Unix 毫秒),向更旧翻页 | + | `after` | int64 | 否 | 0 | **正向增量**(A5):返回 `created_at > after`,**升序**;与 `before` **互斥**(同传 400);负数/非数字回 400 | | `platform` | string | 否 | - | 来源平台过滤(design/16 §4.4)。Web 端不传 = 全量共享主会话;IM 传本平台(如 `telegram`)只拉本渠道消息 | -- **响应**:`200` `{"data":[...]}`,**按时间倒序**(newest first);需 `reverse()` 渲染。 +- **响应**:`200` `{"data":[...]}`;**`before` 时按时间倒序**(newest first,需 `reverse()` 渲染);**`after` 时按时间升序**(最旧在前,可直接顺序渲染/追加)。 `historyMessage` 全字段: | 字段 | 类型 | 说明 | @@ -321,7 +322,7 @@ curl -s http://:8080/api/v1/personas -H "Authorization: Bearer $TOKEN" -H {"data":[{"id":"m1","sessionId":"web","role":"assistant","content":"你好","mode":"normal","mood":"calm","platform":"web","deviceId":"web:chrome:a1b2c3","createdAt":1753...}]} ``` -- **错误**:`400`(`limit`/`before` 非法、`id` 缺失);`501` `{"error":{"code":"not_implemented","message":"会话历史服务未配置(未注入 HistoryStore)"}}`;`500` `{"error":{"code":"messages_failed",...}}`。 +- **错误**:`400`(`limit`/`before`/`after` 非法、`before`+`after` 同传、`id` 缺失);`501` `{"error":{"code":"not_implemented","message":"会话历史服务未配置(未注入 HistoryStore)"}}`;`500` `{"error":{"code":"messages_failed",...}}`。 - **备注**:热分页(design/16 §2.2,§5.1)。`thinking` 只在本 owner/admin 端点可见;WS 回放/广播已剥离。curl: ```bash diff --git a/configs/users/owner.yaml b/configs/users/owner.yaml index 2732ee2..06d2ee0 100644 --- a/configs/users/owner.yaml +++ b/configs/users/owner.yaml @@ -1,32 +1,10 @@ -# 主人档案 / 主人白名单 -# 用途:供 channel 层 actor 元数据 + 策略引擎(design/11 §0/§2)做「认主」判定。 -# 注意:这是策略引擎的引用数据,不是让 LLM/人格读的「身份权限依据」; -# 权限判定绝不由 LLM 读这里得出,而是 channel actor 元数据 + 策略引擎。 -# 后台面板(M10 用户与权限 → 主人档案)可编辑,经 bus/config 热重载供 policy engine(M6)使用。 -# 给 LLM 看的「陪伴对象是谁」身份说明书(人格知识,非权限依据)见 identityFile。 - id: owner displayName: 叶酱 defaultPersona: cyrene -identityFile: configs/users/owner_identity.md - -# 渠道账号白名单:每个渠道确认的主人账号 id(基于渠道协议确认的发送者身份)。 -# 取消注释并填入真实 id 后才生效;未配置时默认「无人被识别为主人」(default deny,安全)。 -# ⚠️ 每个适配器都要单独配,漏配则该平台上的主人消息不会被认出(被最小打扰静默、不回复)。 -# 拿到某平台主人 id 的方式:连上后在该平台发一条消息 → 翻「渠道消息日志」/身份测试, -# 该消息的 deviceId 形如 kook::<你的平台uid>,末尾那段就是 accountId。 accounts: - - platform: kook - accountId: "329284282" - -# 其余适配器:连上并入一条消息后,把上面那段 uid 填进对应 platform(取消注释): -# - platform: qq-obv11 -# accountId: "<主人的 QQ 号 / OneBot uid>" - - platform: weixin-ilink - accountId: "o9cq802QPKPscBenJAp-SMZPBlnQ@im.wechat" -# - platform: telegram -# accountId: "<主人的 Telegram 数字 id>" - -notes: >- - 主人。关系:陪伴对象(昔涟称「开拓者」)。群聊/陌生人使用隔离上下文、记忆只读不写、不泄露主人敏感信息。 - 具体权限(能否读写记忆/用哪些工具/是否代审)由策略引擎基于 actor 元数据决定,与本文件内容无关。 + - platform: kook + accountId: "329284282" + - platform: weixin-ilink + accountId: o9cq802QPKPscBenJAp-SMZPBlnQ@im.wechat +identityFile: configs/users/owner_identity.md +notes: 主人。关系:陪伴对象(昔涟称「开拓者」)。群聊/陌生人使用隔离上下文、记忆只读不写、不泄露主人敏感信息。 具体权限(能否读写记忆/用哪些工具/是否代审)由策略引擎基于 actor 元数据决定,与本文件内容无关。 diff --git a/internal/runtime/history/list_after.go b/internal/runtime/history/list_after.go new file mode 100644 index 0000000..ab80a43 --- /dev/null +++ b/internal/runtime/history/list_after.go @@ -0,0 +1,47 @@ +package history + +import ( + "context" + "fmt" +) + +// ListAfter returns up to limit messages for sessionID strictly newer than after +// (created_at > after) in oldest-first order, for forward incremental catch-up: a +// client that reconnected passes the newest createdAt it has already seen and pulls +// only what it missed, however many messages that is (the "before" cursor only walks +// backwards and cannot recover a gap larger than one page). When platform is non-empty +// the page is further restricted to messages carrying that source platform, exactly as +// in List. rowid breaks created_at ties so messages written in the same millisecond +// keep their insertion order. The returned slice is non-nil; limit is clamped to +// [1, maxPageLimit]. +func (s *Store) ListAfter(ctx context.Context, sessionID string, after int64, limit int, platform string) ([]Message, error) { + limit = clampLimit(limit) + query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at + FROM chat_messages WHERE session_id = ? AND created_at > ?` + args := []any{sessionID, after} + if platform != "" { + query += ` AND platform = ?` + args = append(args, platform) + } + query += ` ORDER BY created_at ASC, rowid ASC LIMIT ?` + args = append(args, limit) + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("history: list after %s: %w", sessionID, err) + } + defer func() { _ = rows.Close() }() + + msgs := make([]Message, 0) + for rows.Next() { + m, err := scanMessage(rows) + if err != nil { + return nil, fmt.Errorf("history: scan list after %s: %w", sessionID, err) + } + msgs = append(msgs, m) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("history: iterate list after %s: %w", sessionID, err) + } + return msgs, nil +} diff --git a/internal/runtime/history/list_after_test.go b/internal/runtime/history/list_after_test.go new file mode 100644 index 0000000..c602872 --- /dev/null +++ b/internal/runtime/history/list_after_test.go @@ -0,0 +1,129 @@ +package history + +import ( + "context" + "testing" +) + +// TestListAfterAscendingExclusive asserts the forward cursor returns only messages +// strictly newer than after, in oldest-first order (the reconnect catch-up direction). +func TestListAfterAscendingExclusive(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + for _, ts := range []int64{1000, 2000, 3000, 4000} { + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "x", CreatedAt: ts}); err != nil { + t.Fatalf("seed %d: %v", ts, err) + } + } + + msgs, err := st.ListAfter(ctx, MainSessionID, 2000, 50, "") + if err != nil { + t.Fatalf("list after: %v", err) + } + if len(msgs) != 2 || msgs[0].CreatedAt != 3000 || msgs[1].CreatedAt != 4000 { + t.Fatalf("after=2000 = %+v, want ascending [3000 4000]", []int64{msgs[0].CreatedAt, msgs[1].CreatedAt}) + } + + // after at the newest message yields an empty (non-nil) page. + msgs, err = st.ListAfter(ctx, MainSessionID, 4000, 50, "") + if err != nil { + t.Fatalf("list after newest: %v", err) + } + if msgs == nil || len(msgs) != 0 { + t.Fatalf("after=4000 = %+v (nil=%v), want empty non-nil", msgs, msgs == nil) + } + + // after=0 is the whole session, still ascending. + msgs, err = st.ListAfter(ctx, MainSessionID, 0, 50, "") + if err != nil { + t.Fatalf("list after 0: %v", err) + } + if len(msgs) != 4 || msgs[0].CreatedAt != 1000 || msgs[3].CreatedAt != 4000 { + t.Fatalf("after=0 = %+v, want ascending [1000 2000 3000 4000]", msgs) + } +} + +// TestListAfterSameTimestampRowIDOrder asserts messages sharing a created_at keep their +// insertion order, so a same-millisecond batch is never reordered or dropped at a page +// boundary (rowid tiebreak). +func TestListAfterSameTimestampRowIDOrder(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + for _, c := range []string{"a", "b", "c"} { + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: c, CreatedAt: 5000}); err != nil { + t.Fatalf("seed %s: %v", c, err) + } + } + msgs, err := st.ListAfter(ctx, MainSessionID, 1000, 50, "") + if err != nil { + t.Fatalf("list after: %v", err) + } + if len(msgs) != 3 || msgs[0].Content != "a" || msgs[1].Content != "b" || msgs[2].Content != "c" { + t.Fatalf("same-ms order = %+v, want [a b c]", []string{msgs[0].Content, msgs[1].Content, msgs[2].Content}) + } +} + +// TestListAfterPlatformFilter asserts the optional platform filter applies to the forward +// query exactly as it does to the backward one. +func TestListAfterPlatformFilter(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + seeds := []Message{ + {SessionID: MainSessionID, Role: RoleUser, Content: "web", Platform: PlatformWeb, CreatedAt: 1000}, + {SessionID: MainSessionID, Role: RoleUser, Content: "tg1", Platform: "telegram", CreatedAt: 2000}, + {SessionID: MainSessionID, Role: RoleUser, Content: "tg2", Platform: "telegram", CreatedAt: 3000}, + } + for _, m := range seeds { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %s: %v", m.Content, err) + } + } + msgs, err := st.ListAfter(ctx, MainSessionID, 1000, 50, "telegram") + if err != nil { + t.Fatalf("list after platform: %v", err) + } + if len(msgs) != 2 || msgs[0].Content != "tg1" || msgs[1].Content != "tg2" { + t.Fatalf("telegram after=1000 = %+v, want [tg1 tg2]", msgs) + } +} + +// TestListAfterLimitClamp asserts a client-supplied limit is capped at maxPageLimit and +// that advancing the cursor pages forward without gaps. +func TestListAfterLimitClamp(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + const total = maxPageLimit + 5 + for i := 0; i < total; i++ { + ts := int64(1000 + i) + if err := st.Append(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "m", CreatedAt: ts}); err != nil { + t.Fatalf("seed %d: %v", ts, err) + } + } + + page, err := st.ListAfter(ctx, MainSessionID, 999, 999, "") + if err != nil { + t.Fatalf("list after clamp: %v", err) + } + if len(page) != maxPageLimit { + t.Fatalf("clamped page len = %d, want %d", len(page), maxPageLimit) + } + if page[0].CreatedAt != 1000 || page[len(page)-1].CreatedAt != 1000+int64(maxPageLimit)-1 { + t.Fatalf("clamped page bounds = [%d..%d]", page[0].CreatedAt, page[len(page)-1].CreatedAt) + } + + // Advance the cursor to the last seen createdAt: the remaining messages, no gap. + next, err := st.ListAfter(ctx, MainSessionID, page[len(page)-1].CreatedAt, 999, "") + if err != nil { + t.Fatalf("list after next: %v", err) + } + if len(next) != total-maxPageLimit { + t.Fatalf("next page len = %d, want %d", len(next), total-maxPageLimit) + } + if next[0].CreatedAt != 1000+int64(maxPageLimit) { + t.Fatalf("next page first = %d, want %d", next[0].CreatedAt, 1000+maxPageLimit) + } +} diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index 6d894f5..242c844 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -132,6 +132,11 @@ type HistoryStore interface { // returned (the Web client requests the full main session, an IM adapter asks for // its own platform only). List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]Message, error) + // ListAfter returns up to limit messages for sessionID strictly newer than after + // (created_at > after) in oldest-first order, so a client that reconnects can pull + // the messages it missed forward from the newest createdAt it has already seen. + // platform filters as in List. limit is clamped to [1, maxPageLimit]. + ListAfter(ctx context.Context, sessionID string, after int64, limit int, platform string) ([]Message, error) // Recent returns the most recent n messages for sessionID in oldest-first order, // for WS replay / context tail usage. Recent(ctx context.Context, sessionID string, n int) ([]Message, error) diff --git a/internal/services/channelbridge/channelbridge_test.go b/internal/services/channelbridge/channelbridge_test.go index 70d0d2e..cb8a244 100644 --- a/internal/services/channelbridge/channelbridge_test.go +++ b/internal/services/channelbridge/channelbridge_test.go @@ -263,6 +263,9 @@ func (c *captureHistory) AppendFull(_ context.Context, msg history.Message) (his func (c *captureHistory) List(context.Context, string, int64, int, string) ([]history.Message, error) { return nil, nil } +func (c *captureHistory) ListAfter(context.Context, string, int64, int, string) ([]history.Message, error) { + return nil, nil +} func (c *captureHistory) Recent(context.Context, string, int) ([]history.Message, error) { return nil, nil } diff --git a/internal/services/gateway/history.go b/internal/services/gateway/history.go index 5d59b77..659a657 100644 --- a/internal/services/gateway/history.go +++ b/internal/services/gateway/history.go @@ -111,9 +111,19 @@ func (s *Server) handleCreateSubSession(w http.ResponseWriter, r *http.Request) writeJSON(w, http.StatusCreated, map[string]any{"sessionId": id}) } -// handleListMessages returns a page of messages for a session, newest first, with cursor -// pagination. It is the hot-paging history endpoint (design/16 §2.2, §5.1): limit defaults -// to 50 (clamped to 200), before is the oldest message's created_at from the previous page. +// handleListMessages returns a page of messages for a session with cursor pagination. +// It is the hot-paging history endpoint (design/16 §2.2, §5.1). Two mutually exclusive +// cursors are supported: +// - before (millisecond createdAt): newest-first page of messages strictly older than +// before — the default, used to page backwards through history; +// - after (millisecond createdAt): oldest-first page of messages strictly newer than +// after — forward incremental catch-up, so a client that reconnected can pull +// everything it missed from the newest createdAt it already has, without the +// one-page limit of a backward walk. +// +// limit defaults to 50 (clamped to 200) for both directions. platform filters the page +// by source platform in either direction. Supplying both before and after (both > 0) +// answers 400, as does a non-numeric/negative after or an invalid limit/before. func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) { if s.history == nil { writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置(未注入 HistoryStore)", HTTPCode: http.StatusNotImplemented}) @@ -143,9 +153,24 @@ func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) { } before = v } + after, ok := parseAfterCursor(r) + if !ok { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid after")) + return + } + if before > 0 && after > 0 { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "before and after are mutually exclusive")) + return + } platform := r.URL.Query().Get("platform") - msgs, err := s.history.List(r.Context(), id, before, limit, platform) + var msgs []history.Message + var err error + if after > 0 { + msgs, err = s.history.ListAfter(r.Context(), id, after, limit, platform) + } else { + msgs, err = s.history.List(r.Context(), id, before, limit, platform) + } if err != nil { s.log.Error("history list messages", "err", err, "session", id) writeError(w, apperr.Wrap(apperr.ErrInternal, "messages_failed", "failed to list messages")) @@ -158,6 +183,21 @@ func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, out) } +// parseAfterCursor reads the non-negative "after" query param (a millisecond createdAt +// cursor). An absent/empty param yields (0, true) — no forward bound. A non-numeric or +// negative value yields (0, false) so the caller answers 400. +func parseAfterCursor(r *http.Request) (int64, bool) { + raw := r.URL.Query().Get("after") + if raw == "" { + return 0, true + } + v, err := strconv.ParseInt(raw, 10, 64) + if err != nil || v < 0 { + return 0, false + } + return v, true +} + // handleChannelPlatformLog returns up to limit messages for an IM platform across its // isolated (group/stranger) sessions, newest first, for the admin adapter log page // (design/16 §4.6 §4). It excludes the fixed main web session, so only the platform's diff --git a/internal/services/gateway/history_after_test.go b/internal/services/gateway/history_after_test.go new file mode 100644 index 0000000..534608d --- /dev/null +++ b/internal/services/gateway/history_after_test.go @@ -0,0 +1,160 @@ +package gateway + +import ( + "context" + "net/http" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// TestHistoryListMessagesAfterCursor covers the forward cursor (A5): after returns the +// strictly-newer messages oldest-first, before keeps its existing newest-first behavior, +// and the two cursors are mutually exclusive. +func TestHistoryListMessagesAfterCursor(t *testing.T) { + t.Parallel() + _, ts, jwt, st := newHistoryEnv(t) + ctx := context.Background() + for _, ms := range []int64{1000, 2000, 3000, 4000} { + if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "x", CreatedAt: ms}); err != nil { + t.Fatalf("seed %d: %v", ms, err) + } + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + base := ts.URL + "/api/v1/sessions/" + history.MainSessionID + "/messages" + + // after=2000 → strictly newer, oldest first. + resp, body := doReq(t, http.MethodGet, base+"?after=2000", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("after status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var forward []historyMessage + decodeDataSlice(t, body, &forward) + if len(forward) != 2 || forward[0].CreatedAt != 3000 || forward[1].CreatedAt != 4000 { + t.Fatalf("after=2000 = %+v, want ascending [3000 4000]", forward) + } + + // after at the newest message → empty page, still 200. + resp, body = doReq(t, http.MethodGet, base+"?after=4000", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("after newest status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var empty []historyMessage + decodeDataSlice(t, body, &empty) + if len(empty) != 0 { + t.Fatalf("after=4000 = %+v, want empty", empty) + } + + // before behavior is unchanged: newest first, strictly older. + resp, body = doReq(t, http.MethodGet, base+"?before=3000&limit=2", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("before status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var backward []historyMessage + decodeDataSlice(t, body, &backward) + if len(backward) != 2 || backward[0].CreatedAt != 2000 || backward[1].CreatedAt != 1000 { + t.Fatalf("before=3000 = %+v, want descending [2000 1000]", backward) + } + + // after=0 is not a forward bound → falls back to the default newest-first page. + resp, body = doReq(t, http.MethodGet, base+"?after=0&limit=2", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("after=0 status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var zero []historyMessage + decodeDataSlice(t, body, &zero) + if len(zero) != 2 || zero[0].CreatedAt != 4000 || zero[1].CreatedAt != 3000 { + t.Fatalf("after=0 = %+v, want default descending [4000 3000]", zero) + } +} + +// TestHistoryListMessagesAfterRejectsBadInput covers the 400s: both cursors together, +// and a negative or non-numeric after. +func TestHistoryListMessagesAfterRejectsBadInput(t *testing.T) { + t.Parallel() + _, ts, jwt, _ := newHistoryEnv(t) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + base := ts.URL + "/api/v1/sessions/" + history.MainSessionID + "/messages" + + for _, q := range []string{ + "?after=1000&before=3000", // both cursors + "?after=-1", // negative + "?after=abc", // non-numeric + } { + resp, body := doReq(t, http.MethodGet, base+q, nil, tok) + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("query %s status = %d, want 400 (body=%s)", q, resp.StatusCode, body) + } + } +} + +// TestHistoryListMessagesAfterPlatformFilter asserts the platform filter composes with +// the forward cursor. +func TestHistoryListMessagesAfterPlatformFilter(t *testing.T) { + t.Parallel() + _, ts, jwt, st := newHistoryEnv(t) + ctx := context.Background() + seeds := []history.Message{ + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "web", Platform: history.PlatformWeb, CreatedAt: 1000}, + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "tg1", Platform: "telegram", CreatedAt: 2000}, + {SessionID: history.MainSessionID, Role: history.RoleUser, Content: "tg2", Platform: "telegram", CreatedAt: 3000}, + } + for _, m := range seeds { + if err := st.Append(ctx, m); err != nil { + t.Fatalf("seed %s: %v", m.Content, err) + } + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodGet, + ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages?after=1000&platform=telegram", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("after+platform status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var tg []historyMessage + decodeDataSlice(t, body, &tg) + if len(tg) != 2 || tg[0].Content != "tg1" || tg[1].Content != "tg2" { + t.Fatalf("after=1000&platform=telegram = %+v, want [tg1 tg2]", tg) + } +} + +// TestHistoryListMessagesAfterLimitClamp asserts the forward page is capped at 200 like +// the backward one, and that advancing the returned cursor pages forward without a gap. +func TestHistoryListMessagesAfterLimitClamp(t *testing.T) { + t.Parallel() + _, ts, jwt, st := newHistoryEnv(t) + ctx := context.Background() + const total = 205 + for i := 0; i < total; i++ { + tsMs := int64(1000 + i) + if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "m", CreatedAt: tsMs}); err != nil { + t.Fatalf("seed %d: %v", tsMs, err) + } + } + tok := signAny(t, jwt, "acct", auth.RoleOwner) + base := ts.URL + "/api/v1/sessions/" + history.MainSessionID + "/messages" + + resp, body := doReq(t, http.MethodGet, base+"?after=999&limit=999", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("clamp status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var page []historyMessage + decodeDataSlice(t, body, &page) + if len(page) != 200 { + t.Fatalf("clamped page len = %d, want 200", len(page)) + } + if page[0].CreatedAt != 1000 || page[len(page)-1].CreatedAt != 1199 { + t.Fatalf("clamped page bounds = [%d..%d], want [1000..1199]", page[0].CreatedAt, page[len(page)-1].CreatedAt) + } + + resp, body = doReq(t, http.MethodGet, base+"?after=1199", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("next page status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var next []historyMessage + decodeDataSlice(t, body, &next) + if len(next) != total-200 || next[0].CreatedAt != 1200 { + t.Fatalf("next page = len %d first %d, want len 5 first 1200", len(next), next[0].CreatedAt) + } +} diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 81f0120..85db8e0 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -114,6 +114,10 @@ type HistoryStore interface { // cross-end realtime sync (design/16 §4.2). AppendFull(ctx context.Context, msg history.Message) (history.Message, error) List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]history.Message, error) + // ListAfter returns up to limit messages strictly newer than after in + // oldest-first order, for forward incremental catch-up after a reconnect + // (the "before" cursor can only walk backwards). platform filters as in List. + ListAfter(ctx context.Context, sessionID string, after int64, limit int, platform string) ([]history.Message, error) Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error) ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]history.Message, error) ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]history.Message, error) diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index d93a1b6..0cfba10 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -179,6 +179,10 @@ func (f *fakeHistory) List(context.Context, string, int64, int, string) ([]histo return nil, nil } +func (f *fakeHistory) ListAfter(context.Context, string, int64, int, string) ([]history.Message, error) { + return nil, nil +} + func (f *fakeHistory) Recent(context.Context, string, int) ([]history.Message, error) { return nil, nil }