Resume gracefully-interrupted subagents and take the RunState repair pins (N3 WP1)
Why: InterruptRunning wrote done.json{status:interrupted} for every
child, and ResumeInterruptedChildren skipped anything with done.json —
graceful shutdowns never resumed (only SIGKILL-style deaths did). done
markers now split by status: terminal states block resume, interrupted
does not. Nested pins carry repairOrphanTail and the N6 fix.
Evidence: go test -count=1 ./... green; TestResumeAfterGracefulInterruptMarker
(resumed once and only once), TestResumeSkipsCompletedMarker (terminal
states still block).
Impact: after any restart, interrupted children with a stored request
re-spawn exactly once regardless of how the process stopped.
Refs: Agent-Native#30, Agent-Squad#34, N3 plan WP1
This commit is contained in:
Vendored
+1
-1
Submodule deps/agent-native updated: 74aa60bf2e...cb21ec4e2a
Vendored
+1
-1
Submodule deps/agent-squad updated: 6306451552...931e5ae410
@@ -138,3 +138,72 @@ func TestSpawnBackgroundRecordsDetach(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// N3 WP1:done.json 语义分流——优雅停机写的 interrupted 不挡续跑(一次且仅一次);
|
||||
// completed 终态挡续跑。
|
||||
func TestResumeAfterGracefulInterruptMarker(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)
|
||||
parentRun := c.run
|
||||
if err := st.CreateRun(&parentRun); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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)
|
||||
// 模拟优雅停机:partial 转录 + done.json{status:"interrupted"}。
|
||||
c.mgr.writeSubagentTranscript(child, []core.Part{{Type: core.PartText, Text: "半截现场"}}, "interrupted")
|
||||
if _, err := st.SealInterrupted(child.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||||
c.mgr.ResumeInterruptedChildren()
|
||||
ids, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||||
if len(ids) != len(before)+1 {
|
||||
t.Fatalf("优雅停机现场应被续跑一次:before=%v after=%v", before, ids)
|
||||
}
|
||||
c.mgr.ResumeInterruptedChildren()
|
||||
ids2, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||||
if len(ids2) != len(ids) {
|
||||
t.Fatalf("重复续跑:%v → %v", ids, ids2)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResumeSkipsCompletedMarker(t *testing.T) {
|
||||
c, st := newSpawnEnv(t, &gateRunner{})
|
||||
c.mgr.runner = noopRunner{}
|
||||
ws := t.TempDir()
|
||||
c.mgr.SetWorkspaceRoot(ws)
|
||||
parentRun := c.run
|
||||
if err := st.CreateRun(&parentRun); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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)
|
||||
c.mgr.writeSubagentTranscript(child, []core.Part{{Type: core.PartText, Text: "已完成"}}, "completed")
|
||||
if _, err := st.SealInterrupted(child.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||||
c.mgr.ResumeInterruptedChildren()
|
||||
ids, _ := st.SubagentRunIDs(c.run.ConversationID)
|
||||
if len(ids) != len(before) {
|
||||
t.Fatalf("completed 终态不得续跑:before=%v after=%v", before, ids)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -86,6 +86,22 @@ func (m *Manager) writeSubagentTranscript(run core.Run, parts []core.Part, statu
|
||||
}
|
||||
}
|
||||
|
||||
// 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"))
|
||||
@@ -116,7 +132,9 @@ func (m *Manager) ResumeInterruptedChildren() {
|
||||
if dir == "" {
|
||||
continue
|
||||
}
|
||||
if fileExists(filepath.Join(dir, "done.json")) || fileExists(filepath.Join(dir, "resumed.json")) {
|
||||
// done.json 语义分流(N3 WP1):终态(completed/failed/cancelled)挡续跑;
|
||||
// interrupted 不挡——优雅停机与崩溃都是可续跑现场。
|
||||
if doneStatus(dir) != "interrupted" || fileExists(filepath.Join(dir, "resumed.json")) {
|
||||
continue
|
||||
}
|
||||
req, err := readSubagentRequest(dir)
|
||||
|
||||
Reference in New Issue
Block a user