feat(memory): P1 长期记忆升级 —— 异步攒批 Consolidate + 软删 + importance/last_seen #1
@@ -23,6 +23,12 @@ func (h *Handler) GenerateReport(c *gin.Context) {
|
|||||||
c.JSON(http.StatusBadRequest, gin.H{"error": "topic required"})
|
c.JSON(http.StatusBadRequest, gin.H{"error": "topic required"})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
// 报告和普通任务一样烧钱,必须过同一道关卡(预算/计费租户/积分硬拦截)。
|
||||||
|
// 此前这里直接 PublishTask,绕过了全部三项。
|
||||||
|
billingTenant, ok := h.preflight(c)
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
id := newReportID()
|
id := newReportID()
|
||||||
graph, _ := json.Marshal(map[string]any{"topic": body.Topic}) // 占位 DSL,报告编排实际读 Meta
|
graph, _ := json.Marshal(map[string]any{"topic": body.Topic}) // 占位 DSL,报告编排实际读 Meta
|
||||||
task := &contract.Task{
|
task := &contract.Task{
|
||||||
@@ -33,10 +39,13 @@ func (h *Handler) GenerateReport(c *gin.Context) {
|
|||||||
contract.MetaTopic: body.Topic,
|
contract.MetaTopic: body.Topic,
|
||||||
contract.MetaKB: body.KB,
|
contract.MetaKB: body.KB,
|
||||||
contract.MetaUserID: userID(c),
|
contract.MetaUserID: userID(c),
|
||||||
|
contract.MetaTenantID: billingTenant, // 用量按计费租户扣,此前报告完全没记 → 漏账
|
||||||
contract.MetaSessionID: sessionID(c),
|
contract.MetaSessionID: sessionID(c),
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
if err := h.bus.PublishTask(c.Request.Context(), task); err != nil {
|
// launch 而非裸 PublishTask:报告也是一次「执行」,要落库(→ 进运行历史、可复盘)
|
||||||
|
// 并开录像(→ SSE 可回放,切走再回来不丢)。
|
||||||
|
if err := h.launch(c, task); err != nil {
|
||||||
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -33,6 +33,56 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blob *blob.Store
|
|||||||
return &Handler{db: db, cache: cache, bus: bus, blob: blob}
|
return &Handler{db: db, cache: cache, bus: bus, blob: blob}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// preflight 是「会烧钱的执行」提交前的统一关卡:当日 token 预算 → 计费租户 → 积分硬拦截。
|
||||||
|
// 返回计费租户;ok=false 表示已写过响应,调用方直接 return。
|
||||||
|
//
|
||||||
|
// 抽出来是因为这套关卡曾经只长在 SubmitTask 上,报告生成(GenerateReport)是另一条路径、
|
||||||
|
// 一直停在最初的「发个 NATS」——于是报告绕过了预算、不记计费租户、余额为 0 也照生成。
|
||||||
|
// 两条路径共用同一个函数,才不会再各长各的。
|
||||||
|
func (h *Handler) preflight(c *gin.Context) (string, bool) {
|
||||||
|
// 成本护栏:单用户当日 token 日预算门控(USER_DAILY_TOKEN_BUDGET,0=不限)。
|
||||||
|
if budget := userDailyTokenBudget(); budget > 0 {
|
||||||
|
uid := userID(c)
|
||||||
|
used := h.cache.GetUsage(c.Request.Context(), uid, time.Now().Format("20060102"))
|
||||||
|
if used >= int64(budget) {
|
||||||
|
c.JSON(http.StatusPaymentRequired, gin.H{
|
||||||
|
"error": "已达当日 token 预算上限", "used": used, "budget": budget,
|
||||||
|
})
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// 计费目标:数据落在活跃租户(工作区),但消耗记到"计费租户"——owner/共享计费→活跃租户,
|
||||||
|
// 否则→本人个人租户(各付各的)。硬拦截与用量都按计费租户走。
|
||||||
|
billingTenant := h.db.ResolveBillingTenantID(c.Request.Context(), userID(c), tenantID(c))
|
||||||
|
// 积分硬拦截(默认关;开关 credit_enforce):计费租户积分余额 ≤0 则拒绝,提示充值。
|
||||||
|
if billingTenant != "" && h.db.CreditEnforceEnabled(c.Request.Context()) {
|
||||||
|
if h.db.TenantBalance(c.Request.Context(), billingTenant) <= 0 {
|
||||||
|
c.JSON(http.StatusPaymentRequired, gin.H{"error": "租户积分余额不足,请充值后再试", "balance_micro": 0})
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return billingTenant, true
|
||||||
|
}
|
||||||
|
|
||||||
|
// launch 把一次执行真正发出去,并接上「执行」该有的全套基建:
|
||||||
|
// 落库(→ 运行历史能看到、能复盘)+ token/轨迹录像(→ SSE 可回放/断点续传,切走再回来不丢)。
|
||||||
|
// 报告生成此前只 PublishTask,这两样都没有,所以报告既进不了运行历史,
|
||||||
|
// 切个页面回来也彻底找不回——它明明在后端好好地跑完了。
|
||||||
|
func (h *Handler) launch(c *gin.Context, task *contract.Task) error {
|
||||||
|
// 持久化任务提交(best-effort:降级模式下静默跳过,不阻断发布)。
|
||||||
|
if err := h.db.SaveTask(c.Request.Context(), userID(c), task.ID, string(task.Graph)); err != nil {
|
||||||
|
log.Printf("[gateway] save task %s failed: %v", task.ID, err)
|
||||||
|
}
|
||||||
|
if err := h.bus.PublishTask(c.Request.Context(), task); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
// 从提交即开始把 token 流 + 执行轨迹录进 Redis Stream(订阅早于 dispatcher 产出)→
|
||||||
|
// SSE 可从中回放/断点续传,根治"连晚/重连丢 token / 丢轨迹事件"。
|
||||||
|
h.startTokenRecorder(task.ID)
|
||||||
|
h.startExecRecorder(task.ID)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// SubmitTask: 解析客户端导出的 JSON DSL,组装为 Task,Publish 到 sundynix.tasks.*。
|
// SubmitTask: 解析客户端导出的 JSON DSL,组装为 Task,Publish 到 sundynix.tasks.*。
|
||||||
func (h *Handler) SubmitTask(c *gin.Context) {
|
func (h *Handler) SubmitTask(c *gin.Context) {
|
||||||
var raw json.RawMessage
|
var raw json.RawMessage
|
||||||
@@ -45,27 +95,10 @@ func (h *Handler) SubmitTask(c *gin.Context) {
|
|||||||
c.JSON(http.StatusUnprocessableEntity, gin.H{"error": err.Error()})
|
c.JSON(http.StatusUnprocessableEntity, gin.H{"error": err.Error()})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// 成本护栏:单用户当日 token 日预算门控(USER_DAILY_TOKEN_BUDGET,0=不限)。超额则拒绝新任务。
|
billingTenant, ok := h.preflight(c)
|
||||||
if budget := userDailyTokenBudget(); budget > 0 {
|
if !ok {
|
||||||
uid := userID(c)
|
|
||||||
used := h.cache.GetUsage(c.Request.Context(), uid, time.Now().Format("20060102"))
|
|
||||||
if used >= int64(budget) {
|
|
||||||
c.JSON(http.StatusPaymentRequired, gin.H{
|
|
||||||
"error": "已达当日 token 预算上限", "used": used, "budget": budget,
|
|
||||||
})
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
|
||||||
// 计费目标:数据落在活跃租户(工作区),但消耗记到"计费租户"——owner/共享计费→活跃租户,
|
|
||||||
// 否则→本人个人租户(各付各的)。硬拦截与用量都按计费租户走。
|
|
||||||
billingTenant := h.db.ResolveBillingTenantID(c.Request.Context(), userID(c), tenantID(c))
|
|
||||||
// 积分硬拦截(默认关;开关 credit_enforce):计费租户积分余额 ≤0 则拒绝新任务,提示充值。
|
|
||||||
if billingTenant != "" && h.db.CreditEnforceEnabled(c.Request.Context()) {
|
|
||||||
if h.db.TenantBalance(c.Request.Context(), billingTenant) <= 0 {
|
|
||||||
c.JSON(http.StatusPaymentRequired, gin.H{"error": "租户积分余额不足,请充值后再试", "balance_micro": 0})
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// 附上用户标识(召回偏好记忆)与会话标识(召回短期多轮历史)。
|
// 附上用户标识(召回偏好记忆)与会话标识(召回短期多轮历史)。
|
||||||
// 真实场景由鉴权/会话中间件注入;此处用请求头,缺省匿名/默认会话。
|
// 真实场景由鉴权/会话中间件注入;此处用请求头,缺省匿名/默认会话。
|
||||||
task.Meta[contract.MetaUserID] = userID(c)
|
task.Meta[contract.MetaUserID] = userID(c)
|
||||||
@@ -75,18 +108,10 @@ func (h *Handler) SubmitTask(c *gin.Context) {
|
|||||||
if c.GetBool("guardrail_suspect") {
|
if c.GetBool("guardrail_suspect") {
|
||||||
task.Meta[contract.MetaSafetyCheck] = true
|
task.Meta[contract.MetaSafetyCheck] = true
|
||||||
}
|
}
|
||||||
// 持久化任务提交(best-effort:降级模式下静默跳过,不阻断发布)。
|
if err := h.launch(c, task); err != nil {
|
||||||
if err := h.db.SaveTask(c.Request.Context(), userID(c), task.ID, string(task.Graph)); err != nil {
|
|
||||||
log.Printf("[gateway] save task %s failed: %v", task.ID, err)
|
|
||||||
}
|
|
||||||
if err := h.bus.PublishTask(c.Request.Context(), task); err != nil {
|
|
||||||
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// 从提交即开始把 token 流 + 执行轨迹录进 Redis Stream(订阅早于 dispatcher 产出)→
|
|
||||||
// SSE 可从中回放/断点续传,根治"连晚/重连丢 token / 丢轨迹事件"。
|
|
||||||
h.startTokenRecorder(task.ID)
|
|
||||||
h.startExecRecorder(task.ID)
|
|
||||||
c.JSON(http.StatusAccepted, gin.H{"task_id": task.ID})
|
c.JSON(http.StatusAccepted, gin.H{"task_id": task.ID})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -96,7 +96,7 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
|
|||||||
p.PUT("/spaces/:id/members/:uid", h.SpaceSetMemberRole) // 改空间成员角色
|
p.PUT("/spaces/:id/members/:uid", h.SpaceSetMemberRole) // 改空间成员角色
|
||||||
p.DELETE("/spaces/:id/members/:uid", h.SpaceRemoveMember) // 移除空间成员
|
p.DELETE("/spaces/:id/members/:uid", h.SpaceRemoveMember) // 移除空间成员
|
||||||
p.POST("/spaces/:id/archive", h.SpaceArchive) // 归档空间
|
p.POST("/spaces/:id/archive", h.SpaceArchive) // 归档空间
|
||||||
p.POST("/reports", h.GenerateReport) // 报告生成
|
p.POST("/reports", middleware.RequireTenantRole(db, store.RoleMember), h.GenerateReport) // 报告生成(同样烧租户积分):viewer 只读拦下
|
||||||
p.GET("/billing", h.Billing)
|
p.GET("/billing", h.Billing)
|
||||||
p.GET("/stats/overview", h.StatsOverview) // 工作台仪表盘聚合
|
p.GET("/stats/overview", h.StatsOverview) // 工作台仪表盘聚合
|
||||||
p.GET("/runs", h.Runs) // 运行历史(复盘)
|
p.GET("/runs", h.Runs) // 运行历史(复盘)
|
||||||
|
|||||||
Reference in New Issue
Block a user