275 lines
9.6 KiB
Go
275 lines
9.6 KiB
Go
package runmgr
|
||
|
||
import (
|
||
"context"
|
||
"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) {
|
||
if err := m.writeSubagentRequestChecked(context.Background(), run, agentID, name, task, hint, detach, nil); err != nil {
|
||
log.Printf("[runmgr] subagent request %s: %v", run.ID, err)
|
||
}
|
||
}
|
||
|
||
// writeSubagentRequestChecked keeps transcript file effects behind an optional
|
||
// live account check. Failures stop later file writes; this auxiliary record
|
||
// cannot be committed atomically with the account database.
|
||
func (m *Manager) writeSubagentRequestChecked(ctx context.Context, run core.Run, agentID, name, task, hint string, detach bool, check func(context.Context) error) error {
|
||
dir := m.subagentDir(run.Owner, run.ConversationID, run.ID)
|
||
if dir == "" {
|
||
return nil
|
||
}
|
||
if err := checkFileEffect(ctx, check); err != nil {
|
||
return err
|
||
}
|
||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||
return err
|
||
}
|
||
body, _ := json.MarshalIndent(subagentRequest{
|
||
AgentID: agentID, Name: name, Task: task, Hint: hint, Detach: detach,
|
||
CreatedAt: time.Now().UTC().Format(time.RFC3339),
|
||
}, "", " ")
|
||
if err := checkFileEffect(ctx, check); err != nil {
|
||
return err
|
||
}
|
||
if err := os.WriteFile(filepath.Join(dir, "request.json"), body, 0o644); err != nil {
|
||
return err
|
||
}
|
||
// 转录文件先建（append 语义清晰可见）。
|
||
if err := checkFileEffect(ctx, check); err != nil {
|
||
return err
|
||
}
|
||
f, err := os.OpenFile(filepath.Join(dir, "transcript.jsonl"), os.O_CREATE|os.O_WRONLY, 0o644)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return f.Close()
|
||
}
|
||
|
||
// writeSubagentTranscript 把本次收集到的 parts 追加进转录并落 done 标记。
|
||
// status 是终态（completed/failed/cancelled/interrupted）。
|
||
func (m *Manager) writeSubagentTranscript(run core.Run, parts []core.Part, status string) {
|
||
if err := m.writeSubagentTranscriptChecked(context.Background(), run, parts, status, nil); err != nil {
|
||
log.Printf("[runmgr] subagent transcript %s: %v", run.ID, err)
|
||
}
|
||
}
|
||
|
||
func (m *Manager) writeSubagentTranscriptChecked(ctx context.Context, run core.Run, parts []core.Part, status string, check func(context.Context) error) error {
|
||
dir := m.subagentDir(run.Owner, run.ConversationID, run.ID)
|
||
if dir == "" {
|
||
return nil
|
||
}
|
||
if err := checkFileEffect(ctx, check); err != nil {
|
||
return err
|
||
}
|
||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||
return err
|
||
}
|
||
if err := checkFileEffect(ctx, check); err != nil {
|
||
return err
|
||
}
|
||
f, err := os.OpenFile(filepath.Join(dir, "transcript.jsonl"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
enc := json.NewEncoder(f)
|
||
for i, p := range parts {
|
||
if err := checkFileEffect(ctx, check); err != nil {
|
||
_ = f.Close()
|
||
return err
|
||
}
|
||
if err := enc.Encode(map[string]any{"index": i, "type": string(p.Type), "text": p.Text, "at": time.Now().UTC().Format(time.RFC3339)}); err != nil {
|
||
_ = f.Close()
|
||
return err
|
||
}
|
||
}
|
||
if err := f.Close(); err != nil {
|
||
return err
|
||
}
|
||
done, _ := json.Marshal(map[string]string{"status": status, "endedAt": time.Now().UTC().Format(time.RFC3339)})
|
||
if err := checkFileEffect(ctx, check); err != nil {
|
||
return err
|
||
}
|
||
if err := os.WriteFile(filepath.Join(dir, "done.json"), done, 0o644); err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// doneStatus 读 done.json 的 status（无文件或不可读按 "interrupted" 处理——
|
||
// 缺标记本身就是中断现场）。
|
||
func doneStatus(dir string) string {
|
||
body, err := os.ReadFile(filepath.Join(dir, "done.json"))
|
||
if err != nil {
|
||
return "interrupted"
|
||
}
|
||
var done struct {
|
||
Status string `json:"status"`
|
||
}
|
||
if json.Unmarshal(body, &done) != nil || done.Status == "" {
|
||
return "interrupted"
|
||
}
|
||
return done.Status
|
||
}
|
||
|
||
// 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() {
|
||
m.ResumeInterruptedChildrenChecked(context.Background(), nil)
|
||
}
|
||
|
||
// ResumeInterruptedChildrenChecked rechecks each run owner before redispatch
|
||
// and before writing the resumed marker. A disable between redispatch and the
|
||
// marker can still leave a spawned run without this file; DB/file atomicity is
|
||
// unavailable, so callers must keep the run admission guard as well.
|
||
func (m *Manager) ResumeInterruptedChildrenChecked(ctx context.Context, check func(context.Context, string) error) {
|
||
// Legacy request files cannot reconstruct a canonical task's attempt, inputs or
|
||
// authority. Preserve its interrupted state rather than silently create a new task.
|
||
records, err := m.st.TaskCards()
|
||
if err != nil {
|
||
log.Printf("[runmgr] resume task bindings: %v (fail-closed)", err)
|
||
return
|
||
}
|
||
bound := make(map[string]bool, len(records))
|
||
for _, record := range records {
|
||
if id := record.Card.Task.RunID; id != "" {
|
||
bound[id] = true
|
||
}
|
||
if id := record.Card.SourceRunID; id != "" {
|
||
bound[id] = true
|
||
}
|
||
}
|
||
runs, err := m.st.ListInterruptedSubagentRuns()
|
||
if err != nil {
|
||
log.Printf("[runmgr] resume scan: %v", err)
|
||
return
|
||
}
|
||
for _, run := range runs {
|
||
if err := ctx.Err(); err != nil {
|
||
log.Printf("[runmgr] resume cancelled: %v", err)
|
||
return
|
||
}
|
||
if bound[run.ID] {
|
||
continue
|
||
}
|
||
// 恢复隔离（协作存储 U3）：未知外部副作用的 run 不自动续跑——hold
|
||
// 未解除即跳过；判定出错同样跳过（fail-closed：宁可人工核对，不盲重派）。
|
||
if held, herr := m.st.HasOpenRecoveryHold(string(run.ID), string(run.ConversationID)); herr != nil || held {
|
||
if herr != nil {
|
||
log.Printf("[runmgr] resume skip %s: recovery hold 判定失败（fail-closed）: %v", run.ID, herr)
|
||
} else {
|
||
log.Printf("[runmgr] resume skip %s: recovery hold 未解除（人工核对后显式释放才会续跑）", run.ID)
|
||
}
|
||
continue
|
||
}
|
||
dir := m.subagentDir(run.Owner, run.ConversationID, run.ID)
|
||
if dir == "" {
|
||
continue
|
||
}
|
||
// done.json 语义分流（N3 WP1）：终态（completed/failed/cancelled）挡续跑；
|
||
// interrupted 不挡——优雅停机与崩溃都是可续跑现场。
|
||
if doneStatus(dir) != "interrupted" || 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") + "，可读取后继续。）"
|
||
if err := checkOwnerFileEffect(ctx, run.Owner, check); err != nil {
|
||
log.Printf("[runmgr] resume skip %s: account check: %v", run.ID, err)
|
||
continue
|
||
}
|
||
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 := checkOwnerFileEffect(ctx, run.Owner, check); err != nil {
|
||
log.Printf("[runmgr] resumed marker skip %s: account check: %v", run.ID, err)
|
||
continue
|
||
}
|
||
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 checkOwnerFileEffect(ctx context.Context, owner string, check func(context.Context, string) error) error {
|
||
if check == nil {
|
||
return checkFileEffect(ctx, nil)
|
||
}
|
||
return checkFileEffect(ctx, func(ctx context.Context) error { return check(ctx, owner) })
|
||
}
|
||
|
||
func fileExists(path string) bool {
|
||
_, err := os.Stat(path)
|
||
return err == nil
|
||
}
|