lyric.tngl.sh / mistermorph
Unified Agent + reusable Go agent core.
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

feat: add observed subtask streaming

Lyric c9e14e18 3c33d21c

+1443 -39
+29
agent/engine_loop.go
··· 507 507 "observation_len", len(item.observation), 508 508 ) 509 509 } 510 + EmitEvent(ctx, nil, Event{ 511 + Kind: EventKindToolDone, 512 + Step: step, 513 + ToolName: strings.TrimSpace(tc.Name), 514 + Status: toolEventStatus(item.err), 515 + Error: eventErrorString(item.err), 516 + }) 510 517 511 518 if item.err == nil { 512 519 if t, ok := e.registry.Get(tc.Name); ok { ··· 626 633 } 627 634 628 635 toolCtx := ctx 636 + EmitEvent(ctx, nil, Event{ 637 + Kind: EventKindToolStart, 638 + ToolName: strings.TrimSpace(tc.Name), 639 + Status: "running", 640 + }) 629 641 if e.subtaskRunner != nil { 630 642 toolCtx = WithSubtaskRunnerContext(toolCtx, e.subtaskRunner) 643 + } 644 + if sink, ok := EventSinkFromContext(ctx); ok { 645 + toolCtx = WithEventSinkContext(toolCtx, sink) 631 646 } 632 647 if e.guard != nil && e.guard.Enabled() && strings.EqualFold(tc.Name, "url_fetch") { 633 648 authProfile, _ := tc.Params["auth_profile"].(string) ··· 684 699 685 700 func normalizedToolName(name string) string { 686 701 return strings.ToLower(strings.TrimSpace(name)) 702 + } 703 + 704 + func toolEventStatus(err error) string { 705 + if err != nil { 706 + return "failed" 707 + } 708 + return "done" 709 + } 710 + 711 + func eventErrorString(err error) string { 712 + if err == nil { 713 + return "" 714 + } 715 + return strings.TrimSpace(err.Error()) 687 716 } 688 717 689 718 func duplicateToolCallObservation(toolName string) string {
+78
agent/events.go
··· 1 + package agent 2 + 3 + import ( 4 + "context" 5 + "strings" 6 + 7 + "github.com/quailyquaily/mistermorph/internal/llmstats" 8 + ) 9 + 10 + const ( 11 + EventKindToolStart = "tool_start" 12 + EventKindToolDone = "tool_done" 13 + EventKindToolOutput = "tool_output" 14 + EventKindSubtaskStart = "subtask_start" 15 + EventKindSubtaskDone = "subtask_done" 16 + ) 17 + 18 + type Event struct { 19 + Kind string `json:"kind"` 20 + RunID string `json:"run_id,omitempty"` 21 + Step int `json:"step,omitempty"` 22 + ToolName string `json:"tool_name,omitempty"` 23 + TaskID string `json:"task_id,omitempty"` 24 + Status string `json:"status,omitempty"` 25 + Mode string `json:"mode,omitempty"` 26 + Profile string `json:"profile,omitempty"` 27 + Stream string `json:"stream,omitempty"` 28 + Text string `json:"text,omitempty"` 29 + Summary string `json:"summary,omitempty"` 30 + OutputKind string `json:"output_kind,omitempty"` 31 + Error string `json:"error,omitempty"` 32 + } 33 + 34 + type EventSink interface { 35 + HandleEvent(context.Context, Event) 36 + } 37 + 38 + type EventSinkFunc func(context.Context, Event) 39 + 40 + func (fn EventSinkFunc) HandleEvent(ctx context.Context, event Event) { 41 + if fn != nil { 42 + fn(ctx, event) 43 + } 44 + } 45 + 46 + type eventSinkContextKey struct{} 47 + 48 + func WithEventSinkContext(ctx context.Context, sink EventSink) context.Context { 49 + if sink == nil { 50 + return ctx 51 + } 52 + if ctx == nil { 53 + ctx = context.Background() 54 + } 55 + return context.WithValue(ctx, eventSinkContextKey{}, sink) 56 + } 57 + 58 + func EventSinkFromContext(ctx context.Context) (EventSink, bool) { 59 + if ctx == nil { 60 + return nil, false 61 + } 62 + sink, ok := ctx.Value(eventSinkContextKey{}).(EventSink) 63 + return sink, ok && sink != nil 64 + } 65 + 66 + func EmitEvent(ctx context.Context, sink EventSink, event Event) { 67 + if sink == nil { 68 + var ok bool 69 + sink, ok = EventSinkFromContext(ctx) 70 + if !ok { 71 + return 72 + } 73 + } 74 + if strings.TrimSpace(event.RunID) == "" { 75 + event.RunID = strings.TrimSpace(llmstats.RunIDFromContext(ctx)) 76 + } 77 + sink.HandleEvent(ctx, event) 78 + }
+20
agent/local_subtask_runner.go
··· 16 16 if r == nil || r.engine == nil { 17 17 return nil, fmt.Errorf("subtask runner is unavailable") 18 18 } 19 + if err := ValidateSubtaskStart(ctx); err != nil { 20 + return FailedSubtaskResult("", err), nil 21 + } 19 22 20 23 taskID, runCtx, meta := PrepareSubtaskContext(ctx, req.Meta) 21 24 log := r.engine.log 25 + EmitEvent(ctx, nil, Event{ 26 + Kind: EventKindSubtaskStart, 27 + TaskID: taskID, 28 + Mode: localSubtaskMode(req), 29 + Profile: string(NormalizeObserveProfile(string(req.ObserveProfile))), 30 + Status: "running", 31 + }) 22 32 if log != nil { 23 33 log.Info("subtask_start", "task_id", taskID, "mode", localSubtaskMode(req), "output_schema", strings.TrimSpace(req.OutputSchema)) 24 34 } ··· 37 47 38 48 if log != nil && result != nil { 39 49 log.Info("subtask_done", "task_id", taskID, "status", result.Status, "output_kind", result.OutputKind) 50 + } 51 + if result != nil { 52 + EmitEvent(ctx, nil, Event{ 53 + Kind: EventKindSubtaskDone, 54 + TaskID: taskID, 55 + Status: strings.TrimSpace(result.Status), 56 + Summary: strings.TrimSpace(result.Summary), 57 + OutputKind: strings.TrimSpace(result.OutputKind), 58 + Error: strings.TrimSpace(result.Error), 59 + }) 40 60 } 41 61 return result, nil 42 62 }
+78
agent/observe.go
··· 1 + package agent 2 + 3 + import ( 4 + "strings" 5 + "time" 6 + ) 7 + 8 + type ObserveProfile string 9 + 10 + const ( 11 + ObserveProfileDefault ObserveProfile = "default" 12 + ObserveProfileLongShell ObserveProfile = "long_shell" 13 + ObserveProfileWebExtract ObserveProfile = "web_extract" 14 + ) 15 + 16 + type ObservePolicy struct { 17 + Profile ObserveProfile 18 + MaxLLMChecks int 19 + MinInterval time.Duration 20 + MinNewBytes int 21 + MinNewEvents int 22 + ForceOnTerminal bool 23 + ForceOnFailure bool 24 + ForceOnPending bool 25 + StreamOutput bool 26 + } 27 + 28 + func NormalizeObserveProfile(raw string) ObserveProfile { 29 + switch ObserveProfile(strings.ToLower(strings.TrimSpace(raw))) { 30 + case ObserveProfileLongShell: 31 + return ObserveProfileLongShell 32 + case ObserveProfileWebExtract: 33 + return ObserveProfileWebExtract 34 + default: 35 + return ObserveProfileDefault 36 + } 37 + } 38 + 39 + func ObservePolicyForProfile(profile ObserveProfile) ObservePolicy { 40 + switch NormalizeObserveProfile(string(profile)) { 41 + case ObserveProfileLongShell: 42 + return ObservePolicy{ 43 + Profile: ObserveProfileLongShell, 44 + MaxLLMChecks: 0, 45 + MinInterval: 2 * time.Second, 46 + MinNewBytes: 256, 47 + MinNewEvents: 0, 48 + ForceOnTerminal: true, 49 + ForceOnFailure: true, 50 + ForceOnPending: true, 51 + StreamOutput: true, 52 + } 53 + case ObserveProfileWebExtract: 54 + return ObservePolicy{ 55 + Profile: ObserveProfileWebExtract, 56 + MaxLLMChecks: 1, 57 + MinInterval: 0, 58 + MinNewBytes: 0, 59 + MinNewEvents: 0, 60 + ForceOnTerminal: true, 61 + ForceOnFailure: true, 62 + ForceOnPending: true, 63 + StreamOutput: false, 64 + } 65 + default: 66 + return ObservePolicy{ 67 + Profile: ObserveProfileDefault, 68 + MaxLLMChecks: 0, 69 + MinInterval: 0, 70 + MinNewBytes: 0, 71 + MinNewEvents: 0, 72 + ForceOnTerminal: true, 73 + ForceOnFailure: true, 74 + ForceOnPending: true, 75 + StreamOutput: false, 76 + } 77 + } 78 + }
+10 -4
agent/spawn_tool.go
··· 49 49 "type": "string", 50 50 "description": "Optional schema identifier for the child task's structured output.", 51 51 }, 52 + "observe_profile": map[string]any{ 53 + "type": "string", 54 + "description": "Optional local observer profile for this child task. Supported values: default, long_shell, web_extract.", 55 + }, 52 56 }, 53 57 "required": []string{"task", "tools"}, 54 58 } ··· 96 100 } 97 101 outputSchema, _ := params["output_schema"].(string) 98 102 outputSchema = strings.TrimSpace(outputSchema) 103 + observeProfile, _ := params["observe_profile"].(string) 99 104 100 105 req := SubtaskRequest{ 101 - Task: task, 102 - Model: model, 103 - OutputSchema: outputSchema, 104 - Registry: subRegistry, 106 + Task: task, 107 + Model: model, 108 + OutputSchema: outputSchema, 109 + ObserveProfile: NormalizeObserveProfile(observeProfile), 110 + Registry: subRegistry, 105 111 } 106 112 107 113 runner := t.deps.Runner
+15 -6
agent/subtask.go
··· 18 18 SubtaskOutputKindJSON = "json" 19 19 20 20 subtaskSummaryMaxChars = 160 21 + subtaskMaxDepth = 1 21 22 ) 22 23 23 24 type SubtaskRequest struct { 24 - Task string 25 - Model string 26 - OutputSchema string 27 - Registry *tools.Registry 28 - Meta map[string]any 29 - RunFunc SubtaskFunc 25 + Task string 26 + Model string 27 + OutputSchema string 28 + ObserveProfile ObserveProfile 29 + Registry *tools.Registry 30 + Meta map[string]any 31 + RunFunc SubtaskFunc 30 32 } 31 33 32 34 type SubtaskResult struct { ··· 85 87 return 0 86 88 } 87 89 return v 90 + } 91 + 92 + func ValidateSubtaskStart(ctx context.Context) error { 93 + if depth := SubtaskDepthFromContext(ctx); depth >= subtaskMaxDepth { 94 + return fmt.Errorf("subtask depth limit reached (max=%d)", subtaskMaxDepth) 95 + } 96 + return nil 88 97 } 89 98 90 99 func BuildSubtaskTask(task string, outputSchema string) string {
+33 -4
agent/subtask_test.go
··· 160 160 } 161 161 162 162 out, err := tool.Execute(context.Background(), map[string]any{ 163 - "task": "fetch something", 164 - "tools": []any{"url_fetch"}, 165 - "model": "gpt-5.4", 166 - "output_schema": "subtask.test.v1", 163 + "task": "fetch something", 164 + "tools": []any{"url_fetch"}, 165 + "model": "gpt-5.4", 166 + "output_schema": "subtask.test.v1", 167 + "observe_profile": "web_extract", 167 168 }) 168 169 if err != nil { 169 170 t.Fatalf("Execute() error = %v", err) ··· 177 178 } 178 179 if runner.req.OutputSchema != "subtask.test.v1" { 179 180 t.Fatalf("runner output schema = %q, want subtask.test.v1", runner.req.OutputSchema) 181 + } 182 + if runner.req.ObserveProfile != ObserveProfileWebExtract { 183 + t.Fatalf("runner observe profile = %q, want %q", runner.req.ObserveProfile, ObserveProfileWebExtract) 180 184 } 181 185 if runner.req.Registry == nil { 182 186 t.Fatal("runner registry is nil") ··· 215 219 t.Fatalf("unexpected final = %#v", final) 216 220 } 217 221 } 222 + 223 + func TestLocalSubtaskRunnerRejectsNestedDepth(t *testing.T) { 224 + engine := New(noopSubtaskClient{}, tools.NewRegistry(), Config{DefaultModel: "gpt-5.2"}, DefaultPromptSpec()) 225 + runner := &localSubtaskRunner{engine: engine} 226 + 227 + ctx := WithSubtaskDepth(context.Background(), 1) 228 + result, err := runner.RunSubtask(ctx, SubtaskRequest{ 229 + RunFunc: func(context.Context) (*SubtaskResult, error) { 230 + t.Fatal("nested subtask should not execute callback") 231 + return nil, nil 232 + }, 233 + }) 234 + if err != nil { 235 + t.Fatalf("RunSubtask() error = %v", err) 236 + } 237 + if result == nil { 238 + t.Fatal("RunSubtask() result is nil") 239 + } 240 + if result.Status != SubtaskStatusFailed { 241 + t.Fatalf("Status = %q, want failed", result.Status) 242 + } 243 + if !strings.Contains(result.Error, "depth limit") { 244 + t.Fatalf("Error = %q, want depth limit", result.Error) 245 + } 246 + }
+8
cmd/mistermorph/consolecmd/local_runtime.go
··· 594 594 } 595 595 596 596 replySink := newConsoleReplySink(r.streamHub, job.TaskID, r.logger) 597 + eventSink := newConsoleEventPreviewSink(r.streamHub, job.TaskID, r.logger) 598 + if bundle := r.currentBundle(); bundle != nil { 599 + eventSink.observer = newConsoleLLMObserver(bundle.taskRuntime, job.Model, r.logger) 600 + } 597 601 streamer := streaming.NewFinalOutputStreamer(streaming.FinalOutputStreamerOptions{ 598 602 Sink: replySink, 599 603 }) ··· 603 607 } 604 608 605 609 runCtx, cancel := context.WithTimeout(workerCtx, job.Timeout) 610 + runCtx = agent.WithEventSinkContext(runCtx, eventSink) 606 611 final, agentCtx, runErr := r.runTask(runCtx, conversationKey, job, onStream) 607 612 contextDeadline := daemonruntime.IsContextDeadline(runCtx, runErr) 608 613 cancel() 609 614 610 615 if runErr != nil { 616 + eventSink.Close() 611 617 displayErr := strings.TrimSpace(outputfmt.FormatErrorForDisplay(runErr)) 612 618 if displayErr == "" { 613 619 displayErr = strings.TrimSpace(runErr.Error()) ··· 620 626 } 621 627 622 628 if pendingID, ok := pendingApprovalID(final); ok { 629 + eventSink.Close() 623 630 if r.streamHub != nil { 624 631 r.streamHub.PublishStatus(job.TaskID, string(daemonruntime.TaskPending)) 625 632 } ··· 637 644 638 645 finishedAt := time.Now().UTC() 639 646 output := strings.TrimSpace(outputfmt.FormatFinalOutput(final)) 647 + eventSink.Close() 640 648 _ = replySink.Finalize(context.Background(), output) 641 649 streamTracker.LogSummary("done") 642 650 r.completeHeartbeatTask(job, heartbeatTaskResultSuccess, nil, finishedAt)
+431
cmd/mistermorph/consolecmd/stream_events.go
··· 1 + package consolecmd 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "log/slog" 7 + "strings" 8 + "sync" 9 + "time" 10 + 11 + "github.com/quailyquaily/mistermorph/agent" 12 + "github.com/quailyquaily/mistermorph/internal/daemonruntime" 13 + ) 14 + 15 + const consoleEventTailMaxChars = 6000 16 + const consoleObserveTimeout = 8 * time.Second 17 + 18 + type consoleObserveRequest struct { 19 + TaskID string 20 + Profile agent.ObserveProfile 21 + Trigger string 22 + Snapshot string 23 + } 24 + 25 + type consoleSemanticObserver interface { 26 + Summarize(context.Context, consoleObserveRequest) (string, error) 27 + } 28 + 29 + type consoleEventPreviewSink struct { 30 + hub *consoleStreamHub 31 + taskID string 32 + logger *slog.Logger 33 + now func() time.Time 34 + observer consoleSemanticObserver 35 + observeTimeout time.Duration 36 + observeCtx context.Context 37 + observeCancel context.CancelFunc 38 + observeWake chan struct{} 39 + 40 + mu sync.Mutex 41 + subtaskLine string 42 + toolLine string 43 + stdoutTail string 44 + stderrTail string 45 + observerSummary string 46 + subtaskProfile agent.ObserveProfile 47 + toolProfile agent.ObserveProfile 48 + pendingNewBytes int 49 + pendingEvents int 50 + lastPublishAt time.Time 51 + seenOutput bool 52 + observeStarted bool 53 + pendingObserve *consoleObserveRequest 54 + observeCalls map[agent.ObserveProfile]int 55 + } 56 + 57 + func newConsoleEventPreviewSink(hub *consoleStreamHub, taskID string, logger *slog.Logger) *consoleEventPreviewSink { 58 + observeCtx, observeCancel := context.WithCancel(context.Background()) 59 + return &consoleEventPreviewSink{ 60 + hub: hub, 61 + taskID: strings.TrimSpace(taskID), 62 + logger: logger, 63 + now: time.Now, 64 + observeTimeout: consoleObserveTimeout, 65 + observeCtx: observeCtx, 66 + observeCancel: observeCancel, 67 + observeWake: make(chan struct{}, 1), 68 + observeCalls: map[agent.ObserveProfile]int{}, 69 + } 70 + } 71 + 72 + func (s *consoleEventPreviewSink) Close() { 73 + if s == nil || s.observeCancel == nil { 74 + return 75 + } 76 + s.observeCancel() 77 + } 78 + 79 + func (s *consoleEventPreviewSink) HandleEvent(_ context.Context, event agent.Event) { 80 + if s == nil || s.hub == nil || strings.TrimSpace(s.taskID) == "" { 81 + return 82 + } 83 + 84 + text, shouldPublish, observeReq := s.consume(event) 85 + if observeReq != nil { 86 + s.enqueueObserve(*observeReq) 87 + } 88 + if !shouldPublish || strings.TrimSpace(text) == "" { 89 + return 90 + } 91 + s.hub.PublishSnapshot(s.taskID, text) 92 + } 93 + 94 + func (s *consoleEventPreviewSink) consume(event agent.Event) (string, bool, *consoleObserveRequest) { 95 + s.mu.Lock() 96 + defer s.mu.Unlock() 97 + 98 + now := time.Now() 99 + if s.now != nil { 100 + now = s.now() 101 + } 102 + activeProfileBefore := s.activeObserveProfileLocked() 103 + changed := false 104 + forcePublish := false 105 + switch strings.TrimSpace(event.Kind) { 106 + case agent.EventKindSubtaskStart: 107 + s.subtaskLine = formatConsoleSubtaskStart(event) 108 + s.subtaskProfile = agent.NormalizeObserveProfile(event.Profile) 109 + s.observerSummary = "" 110 + s.resetOutputTrackingLocked() 111 + changed = true 112 + forcePublish = true 113 + case agent.EventKindSubtaskDone: 114 + s.subtaskLine = formatConsoleSubtaskDone(event) 115 + s.subtaskProfile = agent.ObserveProfileDefault 116 + changed = true 117 + forcePublish = true 118 + case agent.EventKindToolStart: 119 + s.toolLine = formatConsoleToolStart(event) 120 + s.stdoutTail = "" 121 + s.stderrTail = "" 122 + s.observerSummary = "" 123 + s.toolProfile = profileForEvent(event, s.activeObserveProfileLocked()) 124 + s.resetOutputTrackingLocked() 125 + changed = true 126 + forcePublish = true 127 + case agent.EventKindToolDone: 128 + s.toolLine = formatConsoleToolDone(event) 129 + s.toolProfile = agent.ObserveProfileDefault 130 + changed = true 131 + forcePublish = true 132 + case agent.EventKindToolOutput: 133 + switch strings.TrimSpace(event.Stream) { 134 + case "stdout": 135 + s.stdoutTail = appendConsoleTail(s.stdoutTail, event.Text, consoleEventTailMaxChars) 136 + changed = true 137 + case "stderr": 138 + s.stderrTail = appendConsoleTail(s.stderrTail, event.Text, consoleEventTailMaxChars) 139 + changed = true 140 + } 141 + if changed { 142 + s.pendingNewBytes += len(event.Text) 143 + s.pendingEvents++ 144 + if !s.seenOutput && agent.ObservePolicyForProfile(s.activeObserveProfileLocked()).StreamOutput { 145 + s.seenOutput = true 146 + forcePublish = true 147 + } else if !s.seenOutput { 148 + s.seenOutput = true 149 + } 150 + } 151 + } 152 + 153 + if !changed { 154 + return "", false, nil 155 + } 156 + observeReq := s.buildObserveRequestLocked(event, activeProfileBefore) 157 + if !forcePublish && !s.shouldPublishLocked(now) { 158 + return "", false, observeReq 159 + } 160 + s.lastPublishAt = now 161 + s.pendingNewBytes = 0 162 + s.pendingEvents = 0 163 + text := s.renderLocked() 164 + if s.logger != nil && strings.TrimSpace(text) != "" { 165 + s.logger.Debug("console_event_preview_updated", 166 + "task_id", s.taskID, 167 + "kind", strings.TrimSpace(event.Kind), 168 + "tool", strings.TrimSpace(event.ToolName), 169 + "profile", string(s.activeObserveProfileLocked()), 170 + "chars", len(text), 171 + ) 172 + } 173 + return text, true, observeReq 174 + } 175 + 176 + func (s *consoleEventPreviewSink) shouldPublishLocked(now time.Time) bool { 177 + policy := agent.ObservePolicyForProfile(s.activeObserveProfileLocked()) 178 + if !policy.StreamOutput { 179 + return false 180 + } 181 + if policy.MinNewBytes > 0 && s.pendingNewBytes >= policy.MinNewBytes { 182 + return true 183 + } 184 + if policy.MinNewEvents > 0 && s.pendingEvents >= policy.MinNewEvents { 185 + return true 186 + } 187 + if policy.MinInterval > 0 && !s.lastPublishAt.IsZero() && now.Sub(s.lastPublishAt) >= policy.MinInterval { 188 + return true 189 + } 190 + return false 191 + } 192 + 193 + func (s *consoleEventPreviewSink) resetOutputTrackingLocked() { 194 + s.pendingNewBytes = 0 195 + s.pendingEvents = 0 196 + s.seenOutput = false 197 + } 198 + 199 + func (s *consoleEventPreviewSink) activeObserveProfileLocked() agent.ObserveProfile { 200 + if s.toolProfile != "" && s.toolProfile != agent.ObserveProfileDefault { 201 + return s.toolProfile 202 + } 203 + if s.subtaskProfile != "" && s.subtaskProfile != agent.ObserveProfileDefault { 204 + return s.subtaskProfile 205 + } 206 + return agent.ObserveProfileDefault 207 + } 208 + 209 + func (s *consoleEventPreviewSink) buildObserveRequestLocked(event agent.Event, activeProfileBefore agent.ObserveProfile) *consoleObserveRequest { 210 + if s.observer == nil { 211 + return nil 212 + } 213 + profile := profileForEvent(event, activeProfileBefore) 214 + policy := agent.ObservePolicyForProfile(profile) 215 + if policy.MaxLLMChecks <= 0 { 216 + return nil 217 + } 218 + if s.observeCalls[profile] >= policy.MaxLLMChecks { 219 + return nil 220 + } 221 + if !shouldObserveWithLLM(event, policy) { 222 + return nil 223 + } 224 + snapshot := s.renderObserveSnapshotLocked() 225 + if strings.TrimSpace(snapshot) == "" { 226 + return nil 227 + } 228 + s.observeCalls[profile] = s.observeCalls[profile] + 1 229 + return &consoleObserveRequest{ 230 + TaskID: s.taskID, 231 + Profile: profile, 232 + Trigger: strings.TrimSpace(event.Kind), 233 + Snapshot: snapshot, 234 + } 235 + } 236 + 237 + func (s *consoleEventPreviewSink) renderLocked() string { 238 + parts := make([]string, 0, 4) 239 + if line := strings.TrimSpace(s.subtaskLine); line != "" { 240 + parts = append(parts, line) 241 + } 242 + if line := strings.TrimSpace(s.toolLine); line != "" { 243 + parts = append(parts, line) 244 + } 245 + if summary := strings.TrimSpace(s.observerSummary); summary != "" { 246 + parts = append(parts, "summary:\n"+summary) 247 + } 248 + if out := strings.TrimSpace(s.stdoutTail); out != "" { 249 + parts = append(parts, "stdout:\n"+out) 250 + } 251 + if out := strings.TrimSpace(s.stderrTail); out != "" { 252 + parts = append(parts, "stderr:\n"+out) 253 + } 254 + return strings.TrimSpace(strings.Join(parts, "\n\n")) 255 + } 256 + 257 + func (s *consoleEventPreviewSink) renderObserveSnapshotLocked() string { 258 + return s.renderLocked() 259 + } 260 + 261 + func formatConsoleSubtaskStart(event agent.Event) string { 262 + taskID := strings.TrimSpace(event.TaskID) 263 + mode := strings.TrimSpace(event.Mode) 264 + if mode == "" { 265 + mode = "agent" 266 + } 267 + if taskID == "" { 268 + return fmt.Sprintf("[subtask] started (%s)", mode) 269 + } 270 + return fmt.Sprintf("[subtask %s] started (%s)", taskID, mode) 271 + } 272 + 273 + func formatConsoleSubtaskDone(event agent.Event) string { 274 + taskID := strings.TrimSpace(event.TaskID) 275 + status := strings.TrimSpace(event.Status) 276 + if status == "" { 277 + status = "done" 278 + } 279 + base := "[subtask]" 280 + if taskID != "" { 281 + base = fmt.Sprintf("[subtask %s]", taskID) 282 + } 283 + if summary := strings.TrimSpace(event.Summary); summary != "" { 284 + return fmt.Sprintf("%s %s: %s", base, status, summary) 285 + } 286 + if errText := strings.TrimSpace(event.Error); errText != "" { 287 + return fmt.Sprintf("%s %s: %s", base, status, errText) 288 + } 289 + return fmt.Sprintf("%s %s", base, status) 290 + } 291 + 292 + func formatConsoleToolStart(event agent.Event) string { 293 + name := strings.TrimSpace(event.ToolName) 294 + if name == "" { 295 + name = "tool" 296 + } 297 + return fmt.Sprintf("[%s] running", name) 298 + } 299 + 300 + func formatConsoleToolDone(event agent.Event) string { 301 + name := strings.TrimSpace(event.ToolName) 302 + if name == "" { 303 + name = "tool" 304 + } 305 + if errText := strings.TrimSpace(event.Error); errText != "" { 306 + return fmt.Sprintf("[%s] failed: %s", name, daemonruntime.TruncateUTF8(errText, 160)) 307 + } 308 + status := strings.TrimSpace(event.Status) 309 + if status == "" { 310 + status = "done" 311 + } 312 + return fmt.Sprintf("[%s] %s", name, status) 313 + } 314 + 315 + func profileForEvent(event agent.Event, fallback agent.ObserveProfile) agent.ObserveProfile { 316 + if profile := agent.NormalizeObserveProfile(event.Profile); profile != agent.ObserveProfileDefault { 317 + return profile 318 + } 319 + if strings.EqualFold(strings.TrimSpace(event.ToolName), "bash") { 320 + return agent.ObserveProfileLongShell 321 + } 322 + if fallback != "" { 323 + return fallback 324 + } 325 + return agent.ObserveProfileDefault 326 + } 327 + 328 + func shouldObserveWithLLM(event agent.Event, policy agent.ObservePolicy) bool { 329 + if policy.MaxLLMChecks <= 0 { 330 + return false 331 + } 332 + switch strings.TrimSpace(event.Kind) { 333 + case agent.EventKindToolDone: 334 + return true 335 + default: 336 + return false 337 + } 338 + } 339 + 340 + func (s *consoleEventPreviewSink) enqueueObserve(req consoleObserveRequest) { 341 + if s == nil || s.observer == nil { 342 + return 343 + } 344 + s.mu.Lock() 345 + s.pendingObserve = &req 346 + startWorker := !s.observeStarted 347 + if startWorker { 348 + s.observeStarted = true 349 + } 350 + s.mu.Unlock() 351 + 352 + if startWorker { 353 + go s.observeLoop() 354 + } 355 + select { 356 + case s.observeWake <- struct{}{}: 357 + default: 358 + } 359 + } 360 + 361 + func (s *consoleEventPreviewSink) observeLoop() { 362 + for { 363 + select { 364 + case <-s.observeCtx.Done(): 365 + return 366 + case <-s.observeWake: 367 + req := s.takeObserveRequest() 368 + if req == nil { 369 + continue 370 + } 371 + summary, err := s.runObserve(*req) 372 + if err != nil { 373 + if s.logger != nil { 374 + s.logger.Warn("console_event_observer_failed", 375 + "task_id", s.taskID, 376 + "profile", string(req.Profile), 377 + "trigger", req.Trigger, 378 + "error", err.Error(), 379 + ) 380 + } 381 + continue 382 + } 383 + if strings.TrimSpace(summary) == "" { 384 + continue 385 + } 386 + text := s.storeObservedSummary(summary) 387 + if strings.TrimSpace(text) == "" { 388 + continue 389 + } 390 + s.hub.PublishSnapshot(s.taskID, text) 391 + } 392 + } 393 + } 394 + 395 + func (s *consoleEventPreviewSink) takeObserveRequest() *consoleObserveRequest { 396 + s.mu.Lock() 397 + defer s.mu.Unlock() 398 + req := s.pendingObserve 399 + s.pendingObserve = nil 400 + return req 401 + } 402 + 403 + func (s *consoleEventPreviewSink) runObserve(req consoleObserveRequest) (string, error) { 404 + ctx := s.observeCtx 405 + cancel := func() {} 406 + if s.observeTimeout > 0 { 407 + ctx, cancel = context.WithTimeout(ctx, s.observeTimeout) 408 + } 409 + defer cancel() 410 + return s.observer.Summarize(ctx, req) 411 + } 412 + 413 + func (s *consoleEventPreviewSink) storeObservedSummary(summary string) string { 414 + s.mu.Lock() 415 + defer s.mu.Unlock() 416 + s.observerSummary = strings.TrimSpace(summary) 417 + return s.renderLocked() 418 + } 419 + 420 + func appendConsoleTail(existing string, chunk string, maxChars int) string { 421 + combined := existing + chunk 422 + if maxChars <= 0 { 423 + return combined 424 + } 425 + runes := []rune(combined) 426 + if len(runes) <= maxChars { 427 + return combined 428 + } 429 + keep := string(runes[len(runes)-maxChars:]) 430 + return "...\n" + strings.TrimLeft(keep, "\n") 431 + }
+94
cmd/mistermorph/consolecmd/stream_observer.go
··· 1 + package consolecmd 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "io" 7 + "log/slog" 8 + "strings" 9 + 10 + "github.com/quailyquaily/mistermorph/internal/channelruntime/taskruntime" 11 + "github.com/quailyquaily/mistermorph/llm" 12 + ) 13 + 14 + type consoleLLMObserver struct { 15 + runtime *taskruntime.Runtime 16 + model string 17 + logger *slog.Logger 18 + } 19 + 20 + func newConsoleLLMObserver(rt *taskruntime.Runtime, model string, logger *slog.Logger) consoleSemanticObserver { 21 + if rt == nil { 22 + return nil 23 + } 24 + return &consoleLLMObserver{ 25 + runtime: rt, 26 + model: strings.TrimSpace(model), 27 + logger: logger, 28 + } 29 + } 30 + 31 + func (o *consoleLLMObserver) Summarize(ctx context.Context, req consoleObserveRequest) (string, error) { 32 + if o == nil || o.runtime == nil { 33 + return "", fmt.Errorf("console observer runtime unavailable") 34 + } 35 + route, err := o.runtime.ResolveMainRouteForRun() 36 + if err != nil { 37 + return "", err 38 + } 39 + client, err := o.runtime.CreateClientForRoute(route) 40 + if err != nil { 41 + return "", err 42 + } 43 + defer closeConsoleObserverClient(o.logger, client) 44 + 45 + model := strings.TrimSpace(o.model) 46 + if model == "" { 47 + model = strings.TrimSpace(route.ClientConfig.Model) 48 + } 49 + 50 + result, err := client.Chat(ctx, llm.Request{ 51 + Model: model, 52 + Scene: "console.observe", 53 + Messages: []llm.Message{ 54 + { 55 + Role: "system", 56 + Content: "You summarize noisy tool or subtask progress into one short plain-text progress update. " + 57 + "Write at most two short sentences. Do not repeat raw HTML or long logs. Do not use markdown lists.", 58 + }, 59 + { 60 + Role: "user", 61 + Content: buildConsoleObservePrompt(req), 62 + }, 63 + }, 64 + }) 65 + if err != nil { 66 + return "", err 67 + } 68 + return strings.TrimSpace(result.Text), nil 69 + } 70 + 71 + func buildConsoleObservePrompt(req consoleObserveRequest) string { 72 + var b strings.Builder 73 + b.WriteString("profile=") 74 + b.WriteString(strings.TrimSpace(string(req.Profile))) 75 + b.WriteString("\ntrigger=") 76 + b.WriteString(strings.TrimSpace(req.Trigger)) 77 + b.WriteString("\n\nCurrent task preview snapshot:\n") 78 + b.WriteString(strings.TrimSpace(req.Snapshot)) 79 + b.WriteString("\n\nReturn a short user-facing progress update.") 80 + return b.String() 81 + } 82 + 83 + func closeConsoleObserverClient(logger *slog.Logger, client llm.Client) { 84 + if client == nil { 85 + return 86 + } 87 + closer, ok := client.(io.Closer) 88 + if !ok { 89 + return 90 + } 91 + if err := closer.Close(); err != nil && logger != nil { 92 + logger.Warn("console_observer_client_close_failed", "error", err.Error()) 93 + } 94 + }
+172 -1
cmd/mistermorph/consolecmd/streaming_test.go
··· 1 1 package consolecmd 2 2 3 - import "testing" 3 + import ( 4 + "context" 5 + "fmt" 6 + "strings" 7 + "testing" 8 + "time" 9 + 10 + "github.com/quailyquaily/mistermorph/agent" 11 + ) 12 + 13 + type stubConsoleSemanticObserver struct { 14 + summary string 15 + } 16 + 17 + func (s stubConsoleSemanticObserver) Summarize(_ context.Context, _ consoleObserveRequest) (string, error) { 18 + return s.summary, nil 19 + } 4 20 5 21 func TestConsoleStreamHubEvictsDoneFrameWithoutSubscribers(t *testing.T) { 6 22 hub := newConsoleStreamHub() ··· 37 53 t.Fatal("latest entry retained after last subscriber unsubscribed") 38 54 } 39 55 } 56 + 57 + func TestConsoleEventPreviewSinkPublishesBashTail(t *testing.T) { 58 + hub := newConsoleStreamHub() 59 + sink := newConsoleEventPreviewSink(hub, "task-preview", nil) 60 + 61 + sink.HandleEvent(context.Background(), agent.Event{ 62 + Kind: agent.EventKindToolStart, 63 + ToolName: "bash", 64 + Status: "running", 65 + }) 66 + sink.HandleEvent(context.Background(), agent.Event{ 67 + Kind: agent.EventKindToolOutput, 68 + ToolName: "bash", 69 + Stream: "stdout", 70 + Text: "alpha\n", 71 + }) 72 + sink.HandleEvent(context.Background(), agent.Event{ 73 + Kind: agent.EventKindToolOutput, 74 + ToolName: "bash", 75 + Stream: "stderr", 76 + Text: "warn\n", 77 + }) 78 + sink.HandleEvent(context.Background(), agent.Event{ 79 + Kind: agent.EventKindToolDone, 80 + ToolName: "bash", 81 + Status: "done", 82 + }) 83 + 84 + frame, ok := hub.latest["task-preview"] 85 + if !ok { 86 + t.Fatal("expected preview frame") 87 + } 88 + if !strings.Contains(frame.Text, "[bash] done") { 89 + t.Fatalf("frame.Text = %q, want bash done line", frame.Text) 90 + } 91 + if !strings.Contains(frame.Text, "stdout:\nalpha") { 92 + t.Fatalf("frame.Text = %q, want stdout tail", frame.Text) 93 + } 94 + if !strings.Contains(frame.Text, "stderr:\nwarn") { 95 + t.Fatalf("frame.Text = %q, want stderr tail", frame.Text) 96 + } 97 + } 98 + 99 + func TestConsoleEventPreviewSinkLongShellThrottlesOutput(t *testing.T) { 100 + hub := newConsoleStreamHub() 101 + sink := newConsoleEventPreviewSink(hub, "task-throttle", nil) 102 + 103 + now := time.Unix(1000, 0) 104 + sink.now = func() time.Time { return now } 105 + 106 + sink.HandleEvent(context.Background(), agent.Event{ 107 + Kind: agent.EventKindToolStart, 108 + ToolName: "bash", 109 + Status: "running", 110 + }) 111 + startSeq := hub.latest["task-throttle"].Seq 112 + 113 + sink.HandleEvent(context.Background(), agent.Event{ 114 + Kind: agent.EventKindToolOutput, 115 + ToolName: "bash", 116 + Profile: string(agent.ObserveProfileLongShell), 117 + Stream: "stdout", 118 + Text: "a", 119 + }) 120 + firstOutputSeq := hub.latest["task-throttle"].Seq 121 + if firstOutputSeq <= startSeq { 122 + t.Fatalf("first output should publish immediately, startSeq=%d firstOutputSeq=%d", startSeq, firstOutputSeq) 123 + } 124 + 125 + sink.HandleEvent(context.Background(), agent.Event{ 126 + Kind: agent.EventKindToolOutput, 127 + ToolName: "bash", 128 + Profile: string(agent.ObserveProfileLongShell), 129 + Stream: "stdout", 130 + Text: "b", 131 + }) 132 + if hub.latest["task-throttle"].Seq != firstOutputSeq { 133 + t.Fatalf("small incremental output should not publish immediately") 134 + } 135 + 136 + sink.HandleEvent(context.Background(), agent.Event{ 137 + Kind: agent.EventKindToolOutput, 138 + ToolName: "bash", 139 + Profile: string(agent.ObserveProfileLongShell), 140 + Stream: "stdout", 141 + Text: strings.Repeat("x", 300), 142 + }) 143 + if hub.latest["task-throttle"].Seq == firstOutputSeq { 144 + t.Fatalf("large incremental output should trigger publish") 145 + } 146 + } 147 + 148 + func TestConsoleEventPreviewSinkWebExtractSuppressesRawOutput(t *testing.T) { 149 + hub := newConsoleStreamHub() 150 + sink := newConsoleEventPreviewSink(hub, "task-web", nil) 151 + 152 + sink.HandleEvent(context.Background(), agent.Event{ 153 + Kind: agent.EventKindSubtaskStart, 154 + TaskID: "sub_web", 155 + Mode: "agent", 156 + Profile: string(agent.ObserveProfileWebExtract), 157 + Status: "running", 158 + }) 159 + startSeq := hub.latest["task-web"].Seq 160 + 161 + sink.HandleEvent(context.Background(), agent.Event{ 162 + Kind: agent.EventKindToolOutput, 163 + ToolName: "url_fetch", 164 + Stream: "stdout", 165 + Text: "<html>noise</html>", 166 + }) 167 + if hub.latest["task-web"].Seq != startSeq { 168 + t.Fatalf("web_extract raw output should stay suppressed before terminal event") 169 + } 170 + } 171 + 172 + func TestConsoleEventPreviewSinkWebExtractSchedulesObserverSummary(t *testing.T) { 173 + hub := newConsoleStreamHub() 174 + sink := newConsoleEventPreviewSink(hub, "task-observe", nil) 175 + sink.observer = stubConsoleSemanticObserver{summary: "Found candidate article list and narrowed the target."} 176 + defer sink.Close() 177 + 178 + sink.HandleEvent(context.Background(), agent.Event{ 179 + Kind: agent.EventKindSubtaskStart, 180 + TaskID: "sub_web", 181 + Mode: "agent", 182 + Profile: string(agent.ObserveProfileWebExtract), 183 + Status: "running", 184 + }) 185 + sink.HandleEvent(context.Background(), agent.Event{ 186 + Kind: agent.EventKindToolOutput, 187 + ToolName: "url_fetch", 188 + Stream: "stdout", 189 + Text: "<html>noise</html>", 190 + }) 191 + sink.HandleEvent(context.Background(), agent.Event{ 192 + Kind: agent.EventKindToolDone, 193 + ToolName: "url_fetch", 194 + Status: "done", 195 + }) 196 + 197 + deadline := time.Now().Add(2 * time.Second) 198 + for time.Now().Before(deadline) { 199 + frame, ok := hub.latest["task-observe"] 200 + if ok && strings.Contains(frame.Text, "summary:\nFound candidate article list") { 201 + return 202 + } 203 + time.Sleep(10 * time.Millisecond) 204 + } 205 + frame, ok := hub.latest["task-observe"] 206 + if !ok { 207 + t.Fatal("expected observer preview frame") 208 + } 209 + t.Fatalf("observer summary did not appear, latest=%s", fmt.Sprintf("%#v", frame)) 210 + }
+141 -19
docs/feat/feat_20260405_subagent_runtime.md
··· 94 94 - `taskruntime.Run(...)` 创建 engine 时会自动注入 subtask runner。 95 95 - 子任务路径会显式关闭 runtime tool 自动注入，保证工具白名单语义稳定。 96 96 - agent 子任务默认会关闭 `spawn` 工具，避免递归暴露显式子任务入口。 97 + - 子任务深度当前硬编码限制为 `1`： 98 + - 根任务可以启动一层子任务 99 + - 子任务不能再继续启动下一层子任务 97 100 98 101 这意味着： 99 102 ··· 128 131 - snapshot / final / abort 发布 129 132 - 基于 `taskruntime.Run(... OnStream: ...)` 的流式观察 130 133 134 + 当前又补上了一层本地事件预览： 135 + 136 + - task context 里现在可以挂 `EventSink` 137 + - Console Local runtime 会把 tool/subtask 事件汇总成预览文本 138 + - 预览文本继续复用现有 `streamHub` 139 + - 前端暂时不需要理解新的事件协议 140 + 131 141 这意味着： 132 142 133 143 - “任务有状态、有流、有最终结果” 这套范式已经存在。 ··· 135 145 136 146 ### 3.5 `bash` 已支持显式 direct subtask 137 147 138 - `tools/builtin/bash.go` 当前仍然是一次性收集输出，但已经支持显式 `run_in_subtask`： 148 + `tools/builtin/bash.go` 当前已经支持显式 `run_in_subtask`，并且改成了流式读取 stdout/stderr： 139 149 140 150 - 启动命令 141 - - `cmd.Run()` 142 - - 等结束 151 + - `StdoutPipe/StderrPipe` 152 + - 边读边发 `tool_output` 事件 153 + - 最后统一等待命令结束 143 154 - 一次性返回 `stdout/stderr` 144 155 145 156 如果设置 `run_in_subtask=true`，当前行为是： ··· 152 163 所以当前状态是： 153 164 154 165 - 已有显式 `bash -> subtask`。 155 - - 但还没有流式 bash。 156 - - 没有中间状态 157 - - 没法边跑边观察 158 - - 没法做阶段性摘要 166 + - 已有流式 bash 输出事件。 167 + - Console 已经能看到 bash 运行中的文本预览。 168 + - 还没有基于这些事件做第二层 LLM 观察。 159 169 160 170 ### 3.6 `url_fetch` 只是底层原语 161 171 ··· 325 335 Task string 326 336 Model string 327 337 OutputSchema string 338 + ObserveProfile ObserveProfile 328 339 Registry *tools.Registry 329 340 Meta map[string]any 330 341 RunFunc SubtaskFunc ··· 345 356 346 357 - `RunFunc != nil` 表示 direct subtask。 347 358 - `RunFunc == nil` 表示 agent subtask。 359 + - `ObserveProfile` 用来提示本地观察层应该采用哪种节流策略。 348 360 - 一期仍然没有异步句柄。 349 361 350 362 ### 6.2 本地 `SubtaskRunner` ··· 375 387 376 388 observer 的输入应该是事件流，而不是“定时器 + 全量输出”。 377 389 378 - 建议触发条件： 390 + 建议先把触发条件分成两层： 379 391 380 - 1. 状态变化。 381 - 2. 子任务首次产生输出。 382 - 3. 距上次摘要后新增输出超过阈值。 383 - 4. 长时间无新输出，但任务仍在运行。 384 - 5. 子任务进入 `pending`。 385 - 6. 子任务结束或失败。 386 - 7. 子任务主动发出“需要父级帮助”的信号。 392 + 1. 固定触发 393 + 这层不看任务类型，事件来了就触发： 394 + - 状态变化 395 + - 子任务首次产生输出 396 + - 子任务进入 `pending` 397 + - 子任务结束 398 + - 子任务失败 399 + - 即将超时 400 + 401 + 2. 策略触发 402 + 这层才看任务形态： 403 + - 距上次摘要后新增输出超过阈值 404 + - 距上次摘要后经过了一段时间 405 + - 新增事件数超过阈值 406 + - 出现阶段变化，例如“已找到候选链接”“stderr 开始增长” 407 + 408 + 当前建议是： 409 + 410 + - 先做固定触发 + 本地规则观察。 411 + - 不要一开始就把每个事件都送进 LLM。 387 412 388 413 ### 7.2 三层处理 389 414 ··· 396 421 对输出做 ring buffer、去重、截断、按行合并。 397 422 398 423 3. 汇报层 399 - 只在必要时把压缩后的片段交给 LLM，或者直接形成结构化进度。 424 + 先由本地规则直接形成结构化进度；只有必要时才把压缩后的片段交给 LLM。 400 425 401 426 ### 7.3 什么时候才需要 LLM 402 427 ··· 405 430 优先顺序应该是： 406 431 407 432 1. 先做本地压缩。 408 - 2. 本地压缩够用就不调 LLM。 409 - 3. 只有需要语义归纳时才调 LLM。 433 + 2. 先做本地规则判断。 434 + 3. 本地规则够用就不调 LLM。 435 + 4. 只有需要语义归纳时才调 LLM。 436 + 437 + 这一层不要用“全局一个统一阈值”。 438 + 439 + 不同任务形态的观察策略应该不同，建议用 `ObserveProfile`： 440 + 441 + ```go 442 + type ObserveProfile string 443 + 444 + const ( 445 + ObserveProfileDefault ObserveProfile = "default" 446 + ObserveProfileLongShell ObserveProfile = "long_shell" 447 + ObserveProfileWebExtract ObserveProfile = "web_extract" 448 + ) 449 + 450 + type ObservePolicy struct { 451 + Profile ObserveProfile 452 + MaxLLMChecks int 453 + MinInterval time.Duration 454 + MinNewBytes int 455 + MinNewEvents int 456 + ForceOnTerminal bool 457 + ForceOnFailure bool 458 + ForceOnPending bool 459 + } 460 + ``` 461 + 462 + 建议的一期默认策略： 463 + 464 + - `default` 465 + - 中途不调 LLM 466 + - 只在结束 / 失败 / 即将超时时调一次 467 + - `long_shell` 468 + - 每新增一段输出或达到时间阈值后，最多调少量几次 469 + - 适合 `git pull`、长日志、长命令 470 + - `web_extract` 471 + - 不主要按字节阈值触发 472 + - 主要按阶段变化触发 473 + - 适合网页抽取、候选链接筛选 410 474 411 475 例如： 412 476 413 477 - `git pull` 的阶段性输出，本地规则通常就够。 414 478 - 网页抽取任务需要从候选链接里归纳“第三篇文章”，再用 LLM 更合理。 415 479 480 + ### 7.4 当前实现和下一阶段的边界 481 + 482 + 当前代码里，已经有第一层本地观察： 483 + 484 + 现在已有的是： 485 + 486 + - `tool_start` 487 + - `tool_done` 488 + - `tool_output` 489 + - `subtask_start` 490 + - `subtask_done` 491 + - Console Local 对 LLM `final.output` 的流式快照 492 + - Console Local 对 tool/subtask 事件的本地预览汇总 493 + 494 + 这些事件会先进入本地 `EventSink`，然后在 Console Local runtime 里被汇总成预览文本，继续通过现有 `streamHub` 推给前端。 495 + 496 + 当前已经落下的本地观察规则是： 497 + 498 + - `spawn.observe_profile` 可以显式传给子任务。 499 + - `bash.run_in_subtask` 默认使用 `long_shell`。 500 + - `default` 501 + - 只在固定事件上刷新预览，例如 start / done。 502 + - 中途原始输出不会每个 chunk 都刷到前端。 503 + - `long_shell` 504 + - 首段输出会立即刷新一次。 505 + - 后续按 `MinNewBytes` / `MinInterval` 节流刷新。 506 + - `web_extract` 507 + - 当前先压制中途原始输出。 508 + - 当前在 Console Local 里已经可以异步调用第二层观察者，把一次高噪声快照压成短摘要。 509 + 510 + 当前代码里，已经有一个最小的“观察者 LLM”层，但范围很窄： 511 + 512 + - 它只跑在 Console Local 这条观察链路里。 513 + - 它不会把结果喂回父 agent，只会更新面向用户的运行中预览。 514 + - 当前只在 profile 允许时启用；默认主要给 `web_extract` 这类高噪声场景用。 515 + 516 + 还没有的是： 517 + 518 + - 更细的阶段事件，例如网页抽取里的“候选列表已锁定” 519 + - 更完整的 profile 推断 520 + - 除 Console Local 之外的运行时接入 521 + - 更细的观察者预算和防抖策略 522 + 523 + 所以实现顺序应该是： 524 + 525 + 1. 先把事件稳定推出来 526 + 2. 先把本地规则观察做好 527 + 3. 再给高噪声 profile 接最小的观察者 LLM 528 + 4. 最后再扩成更完整的观察者体系 529 + 530 + 当前剩余工作可以压成 4 组： 531 + 532 + 1. 给网页类高噪声任务补阶段事件，而不是只靠原始输出和终态摘要。 533 + 2. 继续补观察者保护，例如更细的节流、终态只触发一次、profile 级去重。 534 + 3. 把目前只在 Console Local 里的观察链路整理成 runtime 可复用部件。 535 + 4. 最后再做异步子任务句柄，例如 `task_get`、`task_wait`、`task_cancel`。 536 + 416 537 --- 417 538 418 539 ## 8) 触发方式 ··· 430 551 431 552 - 不新增 `mode` 参数。 432 553 - 继续保持同步调用。 433 - - 参数是 `task`、`tools`、`model`、`output_schema`。 554 + - 参数是 `task`、`tools`、`model`、`output_schema`、`observe_profile`。 434 555 - `spawn` 在 engine 装配阶段注册，不放进静态 base registry。 435 556 436 557 也就是说，一期 `spawn` 的语义非常简单： ··· 755 876 756 877 - 定时全量回灌 757 878 - 每次事件都调 LLM 879 + - 不区分任务形态就套一个统一阈值 758 880 759 881 ### 12.3 原始高噪声内容泄漏到父级 760 882
+93 -1
docs/feat/feat_20260405_subagent_runtime_impl.md
··· 142 142 - `go test ./agent ./internal/channelruntime/taskruntime ./tools/builtin` 通过 143 143 - `go test ./...` 通过 144 144 145 + ### 2026-04-06 146 + 147 + - 已补上机制级子任务深度限制： 148 + - 当前 `max_subtask_depth` 先硬编码为 `1` 149 + - `localSubtaskRunner.RunSubtask(...)` 会在进入子任务前检查深度 150 + - `taskruntime.Runtime.RunSubtask(...)` 也会做同样检查 151 + - 超过深度时，不会继续执行子任务，而是直接返回失败 envelope 152 + - 已抽出统一事件接口： 153 + - 新增 run-scoped `agent.EventSink` 154 + - 事件通过 context 透传，不再绑在某个具体 runtime 或 tool 上 155 + - 当前已接入的事件类型有： 156 + - `tool_start` 157 + - `tool_done` 158 + - `tool_output` 159 + - `subtask_start` 160 + - `subtask_done` 161 + - 已把 `bash` 改成流式执行： 162 + - 不再只用 `cmd.Run() + bytes.Buffer` 163 + - 改为 `StdoutPipe/StderrPipe` 164 + - stdout/stderr 会边读边通过 `tool_output` 事件上报 165 + - 最终 observation 和 `subtask.bash.result.v1` envelope 结构不变 166 + - 已把 Console Local 观察链路接上： 167 + - `handleTaskJob(...)` 现在会把 `EventSink` 注入到 task context 168 + - Console 新增本地事件预览汇总器 169 + - 该汇总器会把 tool/subtask 事件压成文本快照，继续复用现有 `streamHub` 170 + - 前端不需要改 WebSocket 协议，仍然只消费 `text/status/done` 171 + - 已把最小观察策略模型落下： 172 + - 新增 `ObserveProfile` / `ObservePolicy` 173 + - 当前内置 `default`、`long_shell`、`web_extract` 174 + - `spawn` 已支持显式 `observe_profile` 175 + - `bash.run_in_subtask` 默认使用 `long_shell` 176 + - Console 预览层会按 profile 决定是否立即刷新、按字节阈值刷新，还是压制原始输出 177 + - 已补上最小第二层观察者： 178 + - 观察者只跑在 Console Local 里 179 + - 通过独立异步 worker 调模型，不阻塞 tool 事件线程 180 + - 当前只在 `ObservePolicy.MaxLLMChecks > 0` 的 profile 上启用 181 + - 现在默认只有 `web_extract` 会异步调一次模型，把高噪声快照压成短摘要 182 + - 观察者摘要只更新运行中预览，不会写回父 agent 上下文 183 + - 已补测试： 184 + - 子任务深度限制 185 + - `bash` 流式输出事件 186 + - Console 事件预览汇总 187 + - `spawn.observe_profile` 透传 188 + - `long_shell` 节流 189 + - `web_extract` 抑制原始输出 190 + - `web_extract` 的观察者摘要 191 + - 测试结果： 192 + - `go test ./agent ./tools/builtin ./internal/channelruntime/taskruntime ./cmd/mistermorph/consolecmd` 通过 193 + - `go test ./...` 通过 194 + 145 195 ## 当前实现边界 146 196 147 197 - `spawn` 仍然只支持同步调用，没有 `mode` 参数。 148 198 - `spawn` 是 engine-scoped tool，不是静态 base registry 工具。 149 199 - 还没有异步子任务句柄，也没有 `task_get` / `task_wait` / `task_cancel`。 150 200 - 还没有独立 `ArtifactStore`。 151 - - `bash` 只接了显式 `run_in_subtask`；还没有自动分流，也没有专门的流式 bash 子任务观察器。 201 + - `bash` 已支持流式 stdout/stderr 事件，但还没有自动分流，也没有独立的观察者 LLM。 152 202 - `spawn` 和 direct subtask 现在都依赖统一的 subtask runner 契约，但仍然只有同步调用，没有后台任务句柄。 203 + - Console 已经有最小第二层观察者，但只覆盖受 profile 控制的高噪声场景。 204 + 205 + ## 剩余工作 206 + 207 + 当前主链路已经可用，后续工作主要集中在下面几项： 208 + 209 + ### A. 网页类任务的阶段事件 210 + 211 + - [ ] 给高噪声网页类子任务补更细的阶段事件 212 + - 例如“已拿到首页” 213 + - “已锁定候选列表” 214 + - “已确认第三篇文章” 215 + - [ ] 不把这层逻辑塞进 `url_fetch` 本身 216 + - [ ] 更适合放在网页抽取子任务或复合工具这一层 217 + 218 + ### B. 观察者保护和节流 219 + 220 + - [ ] 继续补观察者预算控制 221 + - 更细的 `MinInterval` 222 + - 终态强制触发但只触发一次 223 + - profile 级并发 / 去重策略 224 + - [ ] 补“超长输出会截断，但流式事件仍可见”的专项测试 225 + - [ ] 评估是否要把 `consoleStreamFrame` 升级成结构化事件协议 226 + 227 + ### C. 运行时复用 228 + 229 + - [ ] 把当前 Console Local 的观察链路整理成可复用部件 230 + - [ ] 评估 Telegram / Slack / LINE / Lark 是否需要接同一套观察面 231 + - [ ] 明确哪些 runtime 只需要第一层本地观察，哪些需要第二层观察者 232 + 233 + ### D. 异步子任务 234 + 235 + - [ ] 设计并实现异步子任务句柄 236 + - `task_get` 237 + - `task_wait` 238 + - `task_cancel` 239 + - [ ] 让异步子任务继续复用现有事件链和 stream hub 240 + 241 + ### E. 可选后续项 242 + 243 + - [ ] 评估是否还需要独立 `ArtifactStore` 244 + - [ ] 评估 `bash` 是否要支持显式之外的自动分流
+42
internal/channelruntime/taskruntime/runtime.go
··· 280 280 if rt == nil { 281 281 return nil, fmt.Errorf("task runtime is nil") 282 282 } 283 + if err := agent.ValidateSubtaskStart(ctx); err != nil { 284 + return agent.FailedSubtaskResult("", err), nil 285 + } 283 286 task := strings.TrimSpace(req.Task) 284 287 if task == "" && req.RunFunc == nil { 285 288 return nil, fmt.Errorf("empty subtask") ··· 294 297 if req.RunFunc != nil { 295 298 mode = "direct" 296 299 } 300 + agent.EmitEvent(ctx, nil, agent.Event{ 301 + Kind: agent.EventKindSubtaskStart, 302 + TaskID: taskID, 303 + Mode: mode, 304 + Profile: string(agent.NormalizeObserveProfile(string(req.ObserveProfile))), 305 + Status: "running", 306 + }) 297 307 logger.Info("subtask_start", "task_id", taskID, "mode", mode, "output_schema", strings.TrimSpace(req.OutputSchema)) 298 308 299 309 if req.RunFunc != nil { ··· 301 311 if err != nil { 302 312 result := agent.FailedSubtaskResult(taskID, err) 303 313 logger.Info("subtask_done", "task_id", taskID, "status", result.Status, "output_kind", result.OutputKind) 314 + agent.EmitEvent(ctx, nil, agent.Event{ 315 + Kind: agent.EventKindSubtaskDone, 316 + TaskID: taskID, 317 + Status: strings.TrimSpace(result.Status), 318 + Summary: strings.TrimSpace(result.Summary), 319 + OutputKind: strings.TrimSpace(result.OutputKind), 320 + Error: strings.TrimSpace(result.Error), 321 + }) 304 322 return result, nil 305 323 } 306 324 result := agent.NormalizeDirectSubtaskResult(taskID, req.OutputSchema, directResult) 307 325 logger.Info("subtask_done", "task_id", taskID, "status", result.Status, "output_kind", result.OutputKind) 326 + agent.EmitEvent(ctx, nil, agent.Event{ 327 + Kind: agent.EventKindSubtaskDone, 328 + TaskID: taskID, 329 + Status: strings.TrimSpace(result.Status), 330 + Summary: strings.TrimSpace(result.Summary), 331 + OutputKind: strings.TrimSpace(result.OutputKind), 332 + Error: strings.TrimSpace(result.Error), 333 + }) 308 334 return result, nil 309 335 } 310 336 ··· 320 346 if err != nil { 321 347 failed := agent.FailedSubtaskResult(taskID, err) 322 348 logger.Info("subtask_done", "task_id", taskID, "status", failed.Status, "output_kind", failed.OutputKind) 349 + agent.EmitEvent(ctx, nil, agent.Event{ 350 + Kind: agent.EventKindSubtaskDone, 351 + TaskID: taskID, 352 + Status: strings.TrimSpace(failed.Status), 353 + Summary: strings.TrimSpace(failed.Summary), 354 + OutputKind: strings.TrimSpace(failed.OutputKind), 355 + Error: strings.TrimSpace(failed.Error), 356 + }) 323 357 return failed, nil 324 358 } 325 359 final := agent.SubtaskResultFromFinal(taskID, req.OutputSchema, result.Final) 326 360 logger.Info("subtask_done", "task_id", taskID, "status", final.Status, "output_kind", final.OutputKind) 361 + agent.EmitEvent(ctx, nil, agent.Event{ 362 + Kind: agent.EventKindSubtaskDone, 363 + TaskID: taskID, 364 + Status: strings.TrimSpace(final.Status), 365 + Summary: strings.TrimSpace(final.Summary), 366 + OutputKind: strings.TrimSpace(final.OutputKind), 367 + Error: strings.TrimSpace(final.Error), 368 + }) 327 369 return final, nil 328 370 } 329 371
+58
internal/channelruntime/taskruntime/subtask_runner_test.go
··· 194 194 t.Fatalf("expected direct subtask to skip llm client, got %d requests", len(client.requests)) 195 195 } 196 196 } 197 + 198 + func TestRunSubtaskRejectsNestedDepth(t *testing.T) { 199 + client := &stubTaskRuntimeClient{} 200 + route := llmutil.ResolvedRoute{ 201 + ClientConfig: llmconfig.ClientConfig{ 202 + Provider: "openai", 203 + Model: "gpt-5.2", 204 + }, 205 + } 206 + rt, err := Bootstrap(depsutil.CommonDependencies{ 207 + Logger: func() (*slog.Logger, error) { 208 + return slog.Default(), nil 209 + }, 210 + LogOptions: func() agent.LogOptions { 211 + return agent.LogOptions{} 212 + }, 213 + ResolveLLMRoute: func(string) (llmutil.ResolvedRoute, error) { 214 + return route, nil 215 + }, 216 + CreateLLMClient: func(llmutil.ResolvedRoute) (llm.Client, error) { 217 + return client, nil 218 + }, 219 + Registry: func() *tools.Registry { 220 + return tools.NewRegistry() 221 + }, 222 + PromptSpec: func(_ context.Context, _ *slog.Logger, _ agent.LogOptions, _ string, _ llm.Client, _ string, _ []string) (agent.PromptSpec, []string, error) { 223 + return agent.DefaultPromptSpec(), nil, nil 224 + }, 225 + }, BootstrapOptions{ 226 + AgentConfig: agent.Config{MaxSteps: 2, ParseRetries: 0, ToolRepeatLimit: 2}, 227 + }) 228 + if err != nil { 229 + t.Fatalf("Bootstrap() error = %v", err) 230 + } 231 + 232 + ctx := agent.WithSubtaskDepth(context.Background(), 1) 233 + result, err := rt.RunSubtask(ctx, agent.SubtaskRequest{ 234 + RunFunc: func(context.Context) (*agent.SubtaskResult, error) { 235 + t.Fatal("nested subtask should not execute callback") 236 + return nil, nil 237 + }, 238 + }) 239 + if err != nil { 240 + t.Fatalf("RunSubtask() error = %v", err) 241 + } 242 + if result == nil { 243 + t.Fatal("RunSubtask() result is nil") 244 + } 245 + if result.Status != agent.SubtaskStatusFailed { 246 + t.Fatalf("Status = %q, want failed", result.Status) 247 + } 248 + if !strings.Contains(result.Error, "depth limit") { 249 + t.Fatalf("Error = %q, want depth limit", result.Error) 250 + } 251 + if len(client.requests) != 0 { 252 + t.Fatalf("nested subtask should not reach llm client, got %d requests", len(client.requests)) 253 + } 254 + }
+74 -4
tools/builtin/bash.go
··· 5 5 "context" 6 6 "encoding/json" 7 7 "fmt" 8 + "io" 8 9 "os" 9 10 "os/exec" 10 11 "path/filepath" 11 12 "strconv" 12 13 "strings" 14 + "sync" 13 15 "time" 14 16 15 17 "github.com/quailyquaily/mistermorph/agent" ··· 141 143 return marshalBashSubtaskResult(result, err) 142 144 } 143 145 req := agent.SubtaskRequest{ 144 - OutputSchema: "subtask.bash.result.v1", 146 + OutputSchema: "subtask.bash.result.v1", 147 + ObserveProfile: agent.ObserveProfileLongShell, 145 148 RunFunc: func(runCtx context.Context) (*agent.SubtaskResult, error) { 146 149 payload, err := t.runCommand(runCtx, cmdStr, cwd, timeout) 147 150 return buildBashSubtaskResult("", payload, err), nil ··· 168 171 var stderr limitedBuffer 169 172 stdout.Limit = t.MaxOutputBytes 170 173 stderr.Limit = t.MaxOutputBytes 171 - cmd.Stdout = &stdout 172 - cmd.Stderr = &stderr 174 + stdoutPipe, err := cmd.StdoutPipe() 175 + if err != nil { 176 + return bashExecutionPayload{}, err 177 + } 178 + stderrPipe, err := cmd.StderrPipe() 179 + if err != nil { 180 + return bashExecutionPayload{}, err 181 + } 182 + if err := cmd.Start(); err != nil { 183 + return bashExecutionPayload{}, err 184 + } 185 + 186 + var streamWG sync.WaitGroup 187 + var stdoutReadErr error 188 + var stderrReadErr error 189 + streamWG.Add(2) 190 + go func() { 191 + defer streamWG.Done() 192 + stdoutReadErr = t.captureCommandStream(runCtx, "stdout", stdoutPipe, &stdout) 193 + }() 194 + go func() { 195 + defer streamWG.Done() 196 + stderrReadErr = t.captureCommandStream(runCtx, "stderr", stderrPipe, &stderr) 197 + }() 173 198 174 - err := cmd.Run() 199 + err = cmd.Wait() 200 + streamWG.Wait() 201 + if err == nil { 202 + if stdoutReadErr != nil { 203 + err = stdoutReadErr 204 + } else if stderrReadErr != nil { 205 + err = stderrReadErr 206 + } 207 + } 175 208 exitCode := 0 176 209 if err != nil { 177 210 switch { ··· 200 233 err = fmt.Errorf("bash exited with code %d", exitCode) 201 234 } 202 235 return payload, err 236 + } 237 + 238 + func (t *BashTool) captureCommandStream(ctx context.Context, stream string, r io.Reader, dst *limitedBuffer) error { 239 + if r == nil || dst == nil { 240 + return nil 241 + } 242 + buf := make([]byte, 4096) 243 + for { 244 + n, err := r.Read(buf) 245 + if n > 0 { 246 + chunk := append([]byte(nil), buf[:n]...) 247 + _, _ = dst.Write(chunk) 248 + text := string(bytes.ToValidUTF8(chunk, []byte("\n[non-utf8 output]\n"))) 249 + if strings.TrimSpace(text) != "" { 250 + agent.EmitEvent(ctx, nil, agent.Event{ 251 + Kind: agent.EventKindToolOutput, 252 + ToolName: t.Name(), 253 + Profile: string(agent.ObserveProfileLongShell), 254 + Stream: stream, 255 + Text: text, 256 + Status: "running", 257 + }) 258 + } 259 + } 260 + if err == nil { 261 + continue 262 + } 263 + if err == io.EOF { 264 + return nil 265 + } 266 + select { 267 + case <-ctx.Done(): 268 + return nil 269 + default: 270 + return err 271 + } 272 + } 203 273 } 204 274 205 275 func formatBashObservation(payload bashExecutionPayload) string {
+58
tools/builtin/bash_test.go
··· 6 6 "os" 7 7 "path/filepath" 8 8 "strings" 9 + "sync" 9 10 "testing" 10 11 "time" 11 12 ··· 20 21 func (s *stubBashSubtaskRunner) RunSubtask(_ context.Context, req agent.SubtaskRequest) (*agent.SubtaskResult, error) { 21 22 s.req = req 22 23 return s.result, nil 24 + } 25 + 26 + type recordingEventSink struct { 27 + mu sync.Mutex 28 + events []agent.Event 29 + } 30 + 31 + func (s *recordingEventSink) HandleEvent(_ context.Context, event agent.Event) { 32 + s.mu.Lock() 33 + defer s.mu.Unlock() 34 + s.events = append(s.events, event) 35 + } 36 + 37 + func (s *recordingEventSink) snapshot() []agent.Event { 38 + s.mu.Lock() 39 + defer s.mu.Unlock() 40 + out := make([]agent.Event, len(s.events)) 41 + copy(out, s.events) 42 + return out 23 43 } 24 44 25 45 func TestContainsTokenBoundary(t *testing.T) { ··· 194 214 if got := normalizeInjectedEnvVarName(tc.in); got != tc.want { 195 215 t.Fatalf("normalizeInjectedEnvVarName(%q) = %q, want %q", tc.in, got, tc.want) 196 216 } 217 + } 218 + } 219 + 220 + func TestBashTool_Execute_EmitsStreamEvents(t *testing.T) { 221 + tool := NewBashTool(true, 5*time.Second, 4096) 222 + sink := &recordingEventSink{} 223 + ctx := agent.WithEventSinkContext(context.Background(), sink) 224 + 225 + out, err := tool.Execute(ctx, map[string]any{ 226 + "cmd": "printf 'alpha\\n'; printf 'beta\\n' 1>&2", 227 + }) 228 + if err != nil { 229 + t.Fatalf("Execute() error = %v (out=%q)", err, out) 230 + } 231 + 232 + events := sink.snapshot() 233 + if len(events) == 0 { 234 + t.Fatal("expected stream events, got none") 235 + } 236 + 237 + var stdoutSeen bool 238 + var stderrSeen bool 239 + for _, event := range events { 240 + if event.Kind != agent.EventKindToolOutput || event.ToolName != "bash" { 241 + continue 242 + } 243 + if event.Stream == "stdout" && strings.Contains(event.Text, "alpha") { 244 + stdoutSeen = true 245 + } 246 + if event.Stream == "stderr" && strings.Contains(event.Text, "beta") { 247 + stderrSeen = true 248 + } 249 + } 250 + if !stdoutSeen { 251 + t.Fatalf("stdout stream event missing, events=%#v", events) 252 + } 253 + if !stderrSeen { 254 + t.Fatalf("stderr stream event missing, events=%#v", events) 197 255 } 198 256 } 199 257
+3
web/vitepress/docs/guide/built-in-tools.md
··· 40 40 Executes local `bash` commands to call existing CLIs, run one-off conversions, execute scripts, or inspect the local environment. 41 41 42 42 - Key limits: can be disabled via `tools.bash.enabled`; restricted by `deny_paths` and internal deny-token rules; child processes inherit only an allowlisted environment. 43 + - Current subtask behavior: accepts `run_in_subtask=true` for an explicit direct subtask boundary; when the current runtime exposes a stream sink, stdout/stderr chunks can be previewed before the command exits. 43 44 44 45 ### `url_fetch` 45 46 ··· 68 69 Starts a subtask with its own context and an explicit tool whitelist. The parent agent blocks until the child finishes and receives a structured result envelope. 69 70 70 71 - Key limits: can be disabled via `tools.spawn.enabled`; the child can use only the tool names passed in `tools`; raw child transcript is not returned to the parent loop by default. 72 + - Current depth limit: the built-in subtask mechanism currently allows only one child layer. A child task cannot spawn another child task. 73 + - Current observer hint: `spawn` accepts an optional `observe_profile` parameter. `default` keeps mid-run previews conservative, `long_shell` is suited to long shell/log output, and `web_extract` suppresses raw noisy output until better stage signals exist. 71 74 72 75 ## Runtime Tools 73 76
+3
web/vitepress/docs/ja/guide/built-in-tools.md
··· 40 40 ローカルの `bash` コマンドを実行します。既存 CLI の利用、一時的な変換処理、スクリプト実行、環境確認に使います。 41 41 42 42 - 主な制約: `tools.bash.enabled` で無効化できます。`deny_paths` と内部 deny-token ルールの制約を受け、子プロセスには許可済み環境変数だけが渡されます。 43 + - 現在の子タスク挙動: `run_in_subtask=true` を明示すると direct subtask 境界で実行されます。runtime が stream sink を持つ場合、stdout/stderr はコマンド終了前にプレビューへ流れます。 43 44 44 45 ### `url_fetch` 45 46 ··· 68 69 独立した文脈と明示的なツールホワイトリストを持つ子タスクを起動します。親 agent は子タスク終了まで同期で待ち、統一された構造化 envelope を受け取ります。 69 70 70 71 - 主な制約: `tools.spawn.enabled` で無効化できます。子タスクが使えるのは引数 `tools` で明示したツールだけです。子タスクの生 transcript はデフォルトで親 loop に戻しません。 72 + - 現在の深さ制限: 組み込み subtask 機構は今のところ 1 層までです。子タスクの中からさらに子タスクを起動することはできません。 73 + - 現在の観測ヒント: `spawn` は任意で `observe_profile` を受け取れます。`default` は実行中プレビューを保守的に扱い、`long_shell` は長いシェル出力やログ向き、`web_extract` は生の高ノイズ出力をいったん抑えます。 71 74 72 75 ## ランタイムツール 73 76
+3
web/vitepress/docs/zh/guide/built-in-tools.md
··· 40 40 执行本地 `bash` 命令，调用已有 CLI、做一次性格式转换、跑脚本或查询系统信息。 41 41 42 42 关键限制：可通过 `tools.bash.enabled` 关闭；会受 `deny_paths` 和内部 deny-token 规则限制；`bash` 启动的子进程只继承白名单环境变量。 43 + 当前子任务行为：支持显式 `run_in_subtask=true`，把命令放进 direct subtask 边界里执行；如果当前 runtime 暴露了流式 sink，stdout/stderr 会在命令结束前先出现在预览流里。 43 44 44 45 ### `url_fetch` 45 46 ··· 68 69 启动一个拥有独立上下文和显式工具白名单的子任务。父 agent 会同步等待子任务结束，并收到统一的结构化 envelope。 69 70 70 71 关键限制：可通过 `tools.spawn.enabled` 关闭；子任务只能使用参数 `tools` 里显式列出的工具；默认不会把子任务原始 transcript 回灌给父 loop。 72 + 当前深度限制：内置 subtask 机制现在只允许一层子任务，子任务里不能再继续启动下一层子任务。 73 + 当前观察提示：`spawn` 支持可选 `observe_profile` 参数。`default` 会保守处理运行中预览，`long_shell` 适合长命令和日志，`web_extract` 会先压制原始高噪声输出，等后续阶段信号或更高层摘要。 71 74 72 75 ## 运行时工具 73 76