Give subagents per-run transcripts, resume, persisted notices, and the Background request field (N2 WP5) #90

Merged
weichen merged 1 commits from feature/n2-subagent-wp5 into main 2026-10-07 14:25:04 +00:00
11 changed files with 479 additions and 4 deletions
+4
View File
@@ -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)
+60
View File
@@ -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)
}
}
+140
View File
@@ -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)
}
}
+156
View File
@@ -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
}
+14
View File
@@ -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
+18
View File
@@ -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 (
+63
View File
@@ -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
}
+20
View File
@@ -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()
}
+2 -2
View File
@@ -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 {