Merge pull request 'Give subagents per-run transcripts, resume, persisted notices, and the Background request field (N2 WP5)' (#90) from feature/n2-subagent-wp5 into main
ci / vet-and-test (push) Successful in 45s
ci / vet-and-test (push) Successful in 45s
Give subagents per-run transcripts, resume, persisted notices (N2 WP5)
This commit was merged in pull request #90.
This commit is contained in:
Vendored
+1
-1
Submodule deps/agent-native updated: 7425b32b9e...74aa60bf2e
Vendored
+1
-1
Submodule deps/agent-squad updated: 9d3faefb97...6306451552
@@ -298,6 +298,8 @@ func Run(opt Options) error {
|
||||
if err := mgr.ReconcileInterrupted(); err != nil {
|
||||
return fmt.Errorf("reconcile interrupted runs: %w", err)
|
||||
}
|
||||
// 后台子代理通知重启重投(N2 WP5):未投递结果装回队列并按空闲语义投递。
|
||||
mgr.LoadPendingNotices()
|
||||
mgr.SetSquadRunner(&runtime.SquadRunner{})
|
||||
|
||||
// 意图闸门:插件意图 overlay → agent-intent 注册表(未装载意图 = 行为回退)。
|
||||
@@ -353,6 +355,8 @@ func Run(opt Options) error {
|
||||
}
|
||||
// fs 预留能力沙箱根:<DataDir>/workspaces/<owner>/<conversationID>/。
|
||||
mgr.SetWorkspaceRoot(filepath.Join(opt.DataDir, "workspaces"))
|
||||
// 中断子代理续跑(N2 WP5):依赖转录目录(workspaceRoot)与派生请求。
|
||||
mgr.ResumeInterruptedChildren()
|
||||
|
||||
// 收件箱任务卡:事件走会话事件管道(持久化/seq/重放);关闭即 T0 基线。
|
||||
cardSvc := cardsvc.New(st, mgr.EmitCardEvent)
|
||||
|
||||
@@ -82,6 +82,7 @@ const maxBackgroundChildren = 3
|
||||
|
||||
// bgNotice 是一条尚未投递给父会话的后台子代理结果。
|
||||
type bgNotice struct {
|
||||
id string // 持久化队列主键(重启重投凭据)
|
||||
owner string
|
||||
name string
|
||||
payload core.AgentResultPayload
|
||||
@@ -1356,6 +1357,8 @@ func (m *Manager) startChild(parentCtx context.Context, parent core.Run, agentID
|
||||
if err := m.st.CreateRun(&child); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
// 独立转录(N2 WP5):派生请求落盘,续跑的重建凭据。
|
||||
m.writeSubagentRequest(child, agentID, name, task, hint, detach)
|
||||
m.emit(parent.ConversationID, parent.ID, core.EvAgentSpawn, core.AgentSpawnPayload{AgentRun: child.ID, Name: name, Task: task})
|
||||
|
||||
base := parentCtx
|
||||
@@ -1421,6 +1424,8 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool
|
||||
m.mu.Lock()
|
||||
subParts := append([]core.Part{}, rc.h.assistant...)
|
||||
m.mu.Unlock()
|
||||
// 独立转录落定(含终态 done 标记,供续跑幂等判定)。
|
||||
m.writeSubagentTranscript(child, subParts, string(status))
|
||||
fullText := partsText(subParts)
|
||||
if len(subParts) > 0 {
|
||||
_ = m.st.AppendMessage(&core.Message{ConversationID: child.ConversationID, RunID: child.ID, Role: core.RoleAssistant, Parts: subParts})
|
||||
@@ -1485,6 +1490,15 @@ func (m *Manager) decBackground(convID string) {
|
||||
}
|
||||
|
||||
func (m *Manager) enqueueNotice(convID string, n bgNotice) {
|
||||
// 先落库(重启重投凭据),再进内存热队列;落库失败不丢内存路径。
|
||||
if n.id == "" {
|
||||
n.id = store.NewID("ntc")
|
||||
}
|
||||
if err := m.st.AppendNotice(store.Notice{
|
||||
ID: n.id, ConversationID: convID, Owner: n.owner, Name: n.name, Payload: n.payload,
|
||||
}); err != nil {
|
||||
log.Printf("[runmgr] notice persist %s: %v", convID, err)
|
||||
}
|
||||
m.noticeMu.Lock()
|
||||
defer m.noticeMu.Unlock()
|
||||
if m.notices == nil {
|
||||
@@ -1493,6 +1507,31 @@ func (m *Manager) enqueueNotice(convID string, n bgNotice) {
|
||||
m.notices[convID] = append(m.notices[convID], n)
|
||||
}
|
||||
|
||||
// LoadPendingNotices 启动时把未投递的通知装回内存队列并尝试投递
|
||||
// (N2 WP5:重启重投)。投递遵循既有空闲语义(忙则等下一轮触发)。
|
||||
func (m *Manager) LoadPendingNotices() {
|
||||
items, err := m.st.ListNotices()
|
||||
if err != nil {
|
||||
log.Printf("[runmgr] notice load: %v", err)
|
||||
return
|
||||
}
|
||||
byConv := map[string]bool{}
|
||||
m.noticeMu.Lock()
|
||||
if m.notices == nil {
|
||||
m.notices = map[string][]bgNotice{}
|
||||
}
|
||||
for _, it := range items {
|
||||
m.notices[it.ConversationID] = append(m.notices[it.ConversationID], bgNotice{
|
||||
id: it.ID, owner: it.Owner, name: it.Name, payload: it.Payload,
|
||||
})
|
||||
byConv[it.ConversationID] = true
|
||||
}
|
||||
m.noticeMu.Unlock()
|
||||
for convID := range byConv {
|
||||
m.deliverNotices(convID)
|
||||
}
|
||||
}
|
||||
|
||||
// deliverNotices 在会话空闲时把全部待投递结果合并成一条 system 消息,并启动新的父回合。
|
||||
// 会话忙、或启动时发现忙,就把结果放回队列。
|
||||
func (m *Manager) deliverNotices(convID string) {
|
||||
@@ -1537,6 +1576,17 @@ func (m *Manager) deliverNotices(convID string) {
|
||||
}
|
||||
m.notices[convID] = append(items, m.notices[convID]...)
|
||||
m.noticeMu.Unlock()
|
||||
return
|
||||
}
|
||||
// 投递成功才删持久化行(失败保留下次重投)。
|
||||
ids := make([]string, 0, len(items))
|
||||
for _, n := range items {
|
||||
if n.id != "" {
|
||||
ids = append(ids, n.id)
|
||||
}
|
||||
}
|
||||
if err := m.st.DeleteNotices(ids); err != nil {
|
||||
log.Printf("[runmgr] notice delete: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1597,6 +1647,16 @@ func (m *Manager) InterruptRunning() {
|
||||
h.cancel()
|
||||
}
|
||||
for _, id := range ids {
|
||||
// 子代理 run 先把已收集的部分转录落定(中断可续跑的现场)。
|
||||
if run, err := m.st.GetRunByID(id); err == nil && run.Kind == core.RunSubagent {
|
||||
m.mu.Lock()
|
||||
var parts []core.Part
|
||||
if h, ok := m.active[id]; ok {
|
||||
parts = append(parts, h.assistant...)
|
||||
}
|
||||
m.mu.Unlock()
|
||||
m.writeSubagentTranscript(run, parts, "interrupted")
|
||||
}
|
||||
m.sealRunInterrupted(id)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
package runmgr
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"agent-system-backend/internal/store"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
// ---- N2 WP5:子代理独立转录、中断续跑、通知持久化 ----
|
||||
|
||||
// 派生即落 request.json;完成时 transcript.jsonl + done.json 落定。
|
||||
func TestSubagentTranscriptFilesWritten(t *testing.T) {
|
||||
c, _ := newSpawnEnv(t, &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "结果文本"}}})
|
||||
ws := t.TempDir()
|
||||
c.mgr.SetWorkspaceRoot(ws)
|
||||
res, err := c.SpawnSubagent("检索", "查资料", "提示语")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dir := filepath.Join(ws, c.run.Owner, c.run.ConversationID, "subagents", res.AgentRun)
|
||||
reqBody, err := os.ReadFile(filepath.Join(dir, "request.json"))
|
||||
if err != nil {
|
||||
t.Fatalf("request.json: %v", err)
|
||||
}
|
||||
if !strings.Contains(string(reqBody), "查资料") {
|
||||
t.Fatalf("request.json 内容:%s", reqBody)
|
||||
}
|
||||
trBody, err := os.ReadFile(filepath.Join(dir, "transcript.jsonl"))
|
||||
if err != nil || !strings.Contains(string(trBody), "结果文本") {
|
||||
t.Fatalf("transcript.jsonl:%v %s", err, trBody)
|
||||
}
|
||||
doneBody, err := os.ReadFile(filepath.Join(dir, "done.json"))
|
||||
if err != nil || !strings.Contains(string(doneBody), "completed") {
|
||||
t.Fatalf("done.json:%v %s", err, doneBody)
|
||||
}
|
||||
}
|
||||
|
||||
// 中断子 run 凭 request.json 续跑:同任务重派、hint 带前次转录路径、
|
||||
// resumed.json 防重复续跑。
|
||||
func TestResumeInterruptedChildrenRespawns(t *testing.T) {
|
||||
runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "续跑结果"}}}
|
||||
c, st := newSpawnEnv(t, runner)
|
||||
c.mgr.runner = noopRunner{}
|
||||
ws := t.TempDir()
|
||||
c.mgr.SetWorkspaceRoot(ws)
|
||||
|
||||
// 父 run 落库(续跑要经 store 取回派生上下文)。
|
||||
parentRun := c.run
|
||||
if err := st.CreateRun(&parentRun); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 造一个中断的子 run 现场(等价于崩溃前的派生)。
|
||||
child := core.Run{
|
||||
ID: store.NewID("run"), ParentRunID: c.run.ID, ConversationID: c.run.ConversationID,
|
||||
ProjectID: c.run.ProjectID, Owner: c.run.Owner, Kind: core.RunSubagent, ProfileID: c.run.ProfileID,
|
||||
Status: core.RunRunning, CreatedAt: time.Now().UTC(),
|
||||
}
|
||||
if err := st.CreateRun(&child); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c.mgr.writeSubagentRequest(child, "agent.sub.explore", "检索", "原始任务", "原始提示", true)
|
||||
if _, err := st.SealInterrupted(child.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
before, err := st.SubagentRunIDs(c.run.ConversationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c.mgr.ResumeInterruptedChildren()
|
||||
|
||||
ids, err := st.SubagentRunIDs(c.run.ConversationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(ids) != len(before)+1 {
|
||||
t.Fatalf("续跑应新增一个子 run:before=%v after=%v", before, ids)
|
||||
}
|
||||
// resumed.json 幂等:第二次调用不再新增。
|
||||
c.mgr.ResumeInterruptedChildren()
|
||||
ids2, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||||
if len(ids2) != len(ids) {
|
||||
t.Fatalf("重复续跑:%v → %v", ids, ids2)
|
||||
}
|
||||
}
|
||||
|
||||
// 通知先落库后投递;模拟重启后 LoadPendingNotices 重投并清库。
|
||||
func TestNoticesPersistAndRedeliverAfterRestart(t *testing.T) {
|
||||
runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "后台结果"}}}
|
||||
c, st := newSpawnEnv(t, runner)
|
||||
c.mgr.runner = noopRunner{}
|
||||
// 会话占线:通知只能入库不能投递。
|
||||
if err := c.mgr.claimConversation(c.run.ConversationID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := c.SpawnBackground("agent.sub.general", "甲", "任务甲", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitUntil(t, func() bool {
|
||||
items, err := st.ListNotices()
|
||||
return err == nil && len(items) == 1
|
||||
})
|
||||
// 模拟重启:同一 store 上的新 Manager 只接通知面(会话仍忙时不应丢)。
|
||||
c.mgr.releaseConversation(c.run.ConversationID)
|
||||
m2 := New(st, c.mgr.bus, nil, c.mgr.profiles, c.mgr.settings, noopRunner{}, runner, nil, nil)
|
||||
m2.LoadPendingNotices()
|
||||
waitUntil(t, func() bool {
|
||||
items, err := st.ListNotices()
|
||||
return err == nil && len(items) == 0
|
||||
})
|
||||
texts := systemTexts(t, c)
|
||||
if len(texts) == 0 || !strings.Contains(texts[len(texts)-1], "后台结果") {
|
||||
t.Fatalf("重投的通知正文缺失:%v", texts)
|
||||
}
|
||||
}
|
||||
|
||||
// 适配层 Background 分支:Request.Background=true → SpawnBackground(语义
|
||||
// 与产品 delegate 的 background 参数一致)。在本包内用 runCtx 直验。
|
||||
func TestSpawnBackgroundRecordsDetach(t *testing.T) {
|
||||
runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "bg"}}}
|
||||
c, _ := newSpawnEnv(t, runner)
|
||||
c.mgr.runner = noopRunner{}
|
||||
id, err := c.SpawnBackground("agent.sub.general", "后台", "任务", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
run, err := c.mgr.st.GetRunByID(id)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if run.Kind != core.RunSubagent || run.Status != core.RunCompleted && run.Status != core.RunRunning {
|
||||
t.Fatalf("run = %+v", run)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
package runmgr
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
// subagent_transcript.go — 子代理独立转录与中断续跑(N2 WP5)。
|
||||
// 每个子 run 一个目录:<workspaceRoot>/<owner>/<conv>/subagents/<runID>/
|
||||
// request.json 派生请求(agentID/name/task/hint/detach),续跑的重建凭据
|
||||
// transcript.jsonl 按 part 追加的转录(完成/中断时落定)
|
||||
// done.json 终态标记(resume 幂等:有它不再续跑)
|
||||
// resumed.json 已续跑标记(指向新 run)
|
||||
|
||||
// subagentDir 返回子 run 的转录目录(未配置 workspaceRoot 时为空,写盘静默跳过)。
|
||||
func (m *Manager) subagentDir(owner, convID, runID string) string {
|
||||
if m.workspaceRoot == "" {
|
||||
return ""
|
||||
}
|
||||
return filepath.Join(m.workspaceRoot, filepath.Clean(owner), filepath.Clean(convID), "subagents", filepath.Clean(runID))
|
||||
}
|
||||
|
||||
// subagentRequest 是 request.json 的内容。
|
||||
type subagentRequest struct {
|
||||
AgentID string `json:"agentId"`
|
||||
Name string `json:"name"`
|
||||
Task string `json:"task"`
|
||||
Hint string `json:"hint,omitempty"`
|
||||
Detach bool `json:"detach"`
|
||||
CreatedAt string `json:"createdAt"`
|
||||
}
|
||||
|
||||
// writeSubagentRequest 在派生时落 request.json 与空转录(失败只记日志,
|
||||
// 不阻断派生——转录是辅助面)。
|
||||
func (m *Manager) writeSubagentRequest(run core.Run, agentID, name, task, hint string, detach bool) {
|
||||
dir := m.subagentDir(run.Owner, run.ConversationID, run.ID)
|
||||
if dir == "" {
|
||||
return
|
||||
}
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
log.Printf("[runmgr] subagent dir %s: %v", run.ID, err)
|
||||
return
|
||||
}
|
||||
body, _ := json.MarshalIndent(subagentRequest{
|
||||
AgentID: agentID, Name: name, Task: task, Hint: hint, Detach: detach,
|
||||
CreatedAt: time.Now().UTC().Format(time.RFC3339),
|
||||
}, "", " ")
|
||||
if err := os.WriteFile(filepath.Join(dir, "request.json"), body, 0o644); err != nil {
|
||||
log.Printf("[runmgr] subagent request %s: %v", run.ID, err)
|
||||
}
|
||||
// 转录文件先建(append 语义清晰可见)。
|
||||
if _, err := os.OpenFile(filepath.Join(dir, "transcript.jsonl"), os.O_CREATE|os.O_WRONLY, 0o644); err != nil {
|
||||
log.Printf("[runmgr] subagent transcript %s: %v", run.ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// writeSubagentTranscript 把本次收集到的 parts 追加进转录并落 done 标记。
|
||||
// status 是终态(completed/failed/cancelled/interrupted)。
|
||||
func (m *Manager) writeSubagentTranscript(run core.Run, parts []core.Part, status string) {
|
||||
dir := m.subagentDir(run.Owner, run.ConversationID, run.ID)
|
||||
if dir == "" {
|
||||
return
|
||||
}
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return
|
||||
}
|
||||
f, err := os.OpenFile(filepath.Join(dir, "transcript.jsonl"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
|
||||
if err != nil {
|
||||
log.Printf("[runmgr] transcript open %s: %v", run.ID, err)
|
||||
return
|
||||
}
|
||||
enc := json.NewEncoder(f)
|
||||
for i, p := range parts {
|
||||
_ = enc.Encode(map[string]any{"index": i, "type": string(p.Type), "text": p.Text, "at": time.Now().UTC().Format(time.RFC3339)})
|
||||
}
|
||||
_ = f.Close()
|
||||
done, _ := json.Marshal(map[string]string{"status": status, "endedAt": time.Now().UTC().Format(time.RFC3339)})
|
||||
if err := os.WriteFile(filepath.Join(dir, "done.json"), done, 0o644); err != nil {
|
||||
log.Printf("[runmgr] done marker %s: %v", run.ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// readSubagentRequest 读 request.json(无文件返回 nil, nil)。
|
||||
func readSubagentRequest(dir string) (*subagentRequest, error) {
|
||||
body, err := os.ReadFile(filepath.Join(dir, "request.json"))
|
||||
if os.IsNotExist(err) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var req subagentRequest
|
||||
if err := json.Unmarshal(body, &req); err != nil {
|
||||
return nil, fmt.Errorf("request.json: %w", err)
|
||||
}
|
||||
return &req, nil
|
||||
}
|
||||
|
||||
// ResumeInterruptedChildren 重启后扫描被中断的子代理 run:有派生请求且
|
||||
// 未完成未续跑的,用原请求重新派生(后台),并把前次转录路径写进 hint
|
||||
// 供模型读取后继续(N2 WP5 的「可恢复」)。幂等:done/resumed 标记跳过。
|
||||
func (m *Manager) ResumeInterruptedChildren() {
|
||||
runs, err := m.st.ListInterruptedSubagentRuns()
|
||||
if err != nil {
|
||||
log.Printf("[runmgr] resume scan: %v", err)
|
||||
return
|
||||
}
|
||||
for _, run := range runs {
|
||||
dir := m.subagentDir(run.Owner, run.ConversationID, run.ID)
|
||||
if dir == "" {
|
||||
continue
|
||||
}
|
||||
if fileExists(filepath.Join(dir, "done.json")) || fileExists(filepath.Join(dir, "resumed.json")) {
|
||||
continue
|
||||
}
|
||||
req, err := readSubagentRequest(dir)
|
||||
if err != nil {
|
||||
log.Printf("[runmgr] resume read %s: %v", run.ID, err)
|
||||
continue
|
||||
}
|
||||
if req == nil {
|
||||
continue // WP5 之前的旧 run 没有重建凭据,保持 interrupted 终态
|
||||
}
|
||||
parent, err := m.st.GetRunByID(run.ParentRunID)
|
||||
if err != nil {
|
||||
log.Printf("[runmgr] resume parent %s: %v", run.ParentRunID, err)
|
||||
continue
|
||||
}
|
||||
hint := req.Hint + "\n(本任务是前次中断后的自动续跑;前次转录在 " +
|
||||
filepath.Join(dir, "transcript.jsonl") + ",可读取后继续。)"
|
||||
newID, err := m.spawnBackground(parent, req.AgentID, req.Name, req.Task, hint)
|
||||
if err != nil {
|
||||
log.Printf("[runmgr] resume spawn %s: %v", run.ID, err)
|
||||
continue
|
||||
}
|
||||
marker, _ := json.Marshal(map[string]string{
|
||||
"newRun": newID,
|
||||
"resumedAt": time.Now().UTC().Format(time.RFC3339),
|
||||
})
|
||||
if err := os.WriteFile(filepath.Join(dir, "resumed.json"), marker, 0o644); err != nil {
|
||||
log.Printf("[runmgr] resumed marker %s: %v", run.ID, err)
|
||||
}
|
||||
log.Printf("[runmgr] resumed subagent %s as %s", run.ID, newID)
|
||||
}
|
||||
}
|
||||
|
||||
func fileExists(path string) bool {
|
||||
_, err := os.Stat(path)
|
||||
return err == nil
|
||||
}
|
||||
@@ -313,6 +313,20 @@ func SubagentFactory(rc Ctx) subagent.Factory {
|
||||
if len(req.Deliverables) > 0 {
|
||||
b.WriteString("\n交付物:" + strings.Join(req.Deliverables, ", "))
|
||||
}
|
||||
// Background(N2 WP5):声明驱动后台派生,语义同产品 delegate 的
|
||||
// background 参数;完成结果经通知队列回投,本次调用立即返回 run ID。
|
||||
// rc 无后台能力(测试桩/旧宿主)时回落同步派生,不 fail-closed。
|
||||
if req.Background {
|
||||
if bg, ok := rc.(interface {
|
||||
SpawnBackground(agentID, name, task, hint string) (core.ID, error)
|
||||
}); ok {
|
||||
id, err := bg.SpawnBackground("delegatee", "delegatee", b.String(), "内核 delegate_task 委派(后台)")
|
||||
if err != nil {
|
||||
return subagent.Result{}, err
|
||||
}
|
||||
return subagent.Result{Success: true, Text: "已在后台启动 " + id + ",完成后会自动通知"}, nil
|
||||
}
|
||||
}
|
||||
payload, err := rc.SpawnSubagent("delegatee", b.String(), "内核 delegate_task 委派")
|
||||
if err != nil {
|
||||
return subagent.Result{}, err
|
||||
|
||||
@@ -404,9 +404,27 @@ func init() {
|
||||
migration{version: 16, name: "secrets_context_and_variants",
|
||||
up: map[string][]string{"sqlite": v16Up, "postgres": v16Up},
|
||||
down: map[string][]string{"sqlite": v16Down, "postgres": v16Down}},
|
||||
migration{version: 17, name: "pending_notices",
|
||||
up: map[string][]string{"sqlite": v17Up, "postgres": v17Up},
|
||||
down: map[string][]string{"sqlite": v17Down, "postgres": v17Down}},
|
||||
)
|
||||
}
|
||||
|
||||
// v17 后台子代理完成通知的持久化队列(N2 WP5:重启后重投未投递结果)。
|
||||
var v17Up = []string{
|
||||
`CREATE TABLE IF NOT EXISTS pending_notices (
|
||||
id TEXT PRIMARY KEY,
|
||||
conversation_id TEXT NOT NULL,
|
||||
owner TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
payload TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL
|
||||
)`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_pending_notices_conv ON pending_notices(conversation_id, created_at)`,
|
||||
}
|
||||
|
||||
var v17Down = []string{`DROP TABLE IF EXISTS pending_notices`}
|
||||
|
||||
// v16 密钥独立表、设置快照、条件注入、用户命令、会话要求、消息排除与候选回复。
|
||||
var v16Up = []string{
|
||||
`CREATE TABLE IF NOT EXISTS secret_keys (
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
// Notice 是一条待投递的后台子代理完成通知(N2 WP5)。
|
||||
type Notice struct {
|
||||
ID string
|
||||
ConversationID string
|
||||
Owner string
|
||||
Name string
|
||||
Payload core.AgentResultPayload
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// AppendNotice 落一条待投递通知(重启后可重投)。
|
||||
func (s *Store) AppendNotice(n Notice) error {
|
||||
_, err := s.db.Exec(s.q(`INSERT INTO pending_notices (id,conversation_id,owner,name,payload,created_at) VALUES(?,?,?,?,?,?)`),
|
||||
n.ID, n.ConversationID, n.Owner, n.Name, marshalJSON(&n.Payload), nowISO())
|
||||
return err
|
||||
}
|
||||
|
||||
// ListNotices 取回全部待投递通知(Manager 启动时装入内存队列)。
|
||||
func (s *Store) ListNotices() ([]Notice, error) {
|
||||
rows, err := s.db.Query(s.q(`SELECT id,conversation_id,owner,name,payload,created_at FROM pending_notices ORDER BY created_at`))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Notice
|
||||
for rows.Next() {
|
||||
var n Notice
|
||||
var payload, ca string
|
||||
if err := rows.Scan(&n.ID, &n.ConversationID, &n.Owner, &n.Name, &payload, &ca); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
n.Payload, _ = unmarshalJSON[core.AgentResultPayload](payload)
|
||||
n.CreatedAt, _ = parseTime(ca)
|
||||
out = append(out, n)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// DeleteNotice 投递成功后删除(按 id 幂等)。
|
||||
func (s *Store) DeleteNotice(id string) error {
|
||||
_, err := s.db.Exec(s.q(`DELETE FROM pending_notices WHERE id=?`), id)
|
||||
return err
|
||||
}
|
||||
|
||||
// DeleteNotices 批量删除(投递成功批)。
|
||||
func (s *Store) DeleteNotices(ids []string) error {
|
||||
for _, id := range ids {
|
||||
if err := s.DeleteNotice(id); err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -195,3 +195,23 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message) (bool, error) {
|
||||
}
|
||||
return n == 1, nil
|
||||
}
|
||||
|
||||
// ListInterruptedSubagentRuns 列出被收成 interrupted 的子代理 run
|
||||
// (N2 WP5 重启续跑的候选集;幂等过滤由转录目录的 done/resumed 标记完成)。
|
||||
func (s *Store) ListInterruptedSubagentRuns() ([]core.Run, error) {
|
||||
rows, err := s.db.Query(s.q(`SELECT `+runColumns+` FROM runs WHERE kind=? AND status=? ORDER BY created_at`),
|
||||
string(core.RunSubagent), string(core.RunInterrupted))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []core.Run{}
|
||||
for rows.Next() {
|
||||
r, err := scanRun(rows.Scan)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, r)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
@@ -71,8 +71,8 @@ func TestMigrationUpgradesExistingDatabase(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 16 {
|
||||
t.Fatalf("expected versions [1..16], got %v", got)
|
||||
if len(got) != 17 {
|
||||
t.Fatalf("expected versions [1..17], got %v", got)
|
||||
}
|
||||
// 老库形态(v2 前):events 全局自增、settings 单列主键。
|
||||
if err := st.MigrateDown(1); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user