fix(edge): 恢复主运行 goroutine 与聚合器 panic，幂等 Bus.Close

edge-server/internal/events/bus.go

@@ -96,6 +96,10 @@ type Bus struct {
 	stopCh    chan struct{}
 	jobs      chan observerJob
 	workersWg sync.WaitGroup
+
+	// closeOnce makes Close idempotent: a second call (e.g. from a shutdown
+	// hook + a defer in the same process) must not re-close stopCh and panic.
+	closeOnce sync.Once
 }
 
 // NewBus creates a new event bus with the given maximum history size.
@@ -431,13 +435,19 @@ func (b *Bus) DroppedCount() int64 {
 // Close shuts down the observer worker pool, closes the job channel, and
 // flushes and closes the underlying event log if one was configured via
 // WithEventLogPath. It is safe to call Close on a Bus that has no event log.
+//
+// Close is idempotent: a second call (e.g. from a shutdown hook plus a defer
+// in the same process) returns nil without re-closing stopCh and panicking.
 func (b *Bus) Close() error {
-	// Shut down the observer worker pool.
-	close(b.stopCh)
-	b.workersWg.Wait()
-
-	if b.eventLog != nil {
-		return b.eventLog.Close()
-	}
-	return nil
+	var eventLogErr error
+	b.closeOnce.Do(func() {
+		// Shut down the observer worker pool.
+		close(b.stopCh)
+		b.workersWg.Wait()
+
+		if b.eventLog != nil {
+			eventLogErr = b.eventLog.Close()
+		}
+	})
+	return eventLogErr
 }

edge-server/internal/events/bus_behavior_test.go

@@ -625,7 +625,18 @@ func TestBusCloseShutsDownWorkers(t *testing.T) {
 	// After close, verify the bus is no longer operational.
 	// Worker pool is stopped; observers will not be processed.
 	// Note: Close() closes stopCh and waits for workers, then closes eventLog if any.
-	// It is NOT idempotent — calling Close() twice panics because stopCh is re-closed.
+	// Close() is idempotent via sync.Once — calling it twice is safe (see TestBusCloseIdempotent).
+}
+
+func TestBusCloseIdempotent(t *testing.T) {
+	b := NewBus(10)
+	if err := b.Close(); err != nil {
+		t.Fatalf("first Close: %v", err)
+	}
+	// A second Close must not panic on re-closing stopCh.
+	if err := b.Close(); err != nil {
+		t.Fatalf("second Close: %v", err)
+	}
 }
 
 // ---------------------------------------------------------------------------

edge-server/internal/lifecycle/process_executor_build.go

@@ -130,7 +130,7 @@ func (e *ProcessExecutor) buildAndStartProcess(
 		e.mu.Lock()
 		e.processes[run.ID] = cmd.Process
 		e.mu.Unlock()
-		go e.watchRunProcess(ctx, run.ID, cmd.Process, watchStop)
+		safeGo("watchRunProcess", func() { e.watchRunProcess(ctx, run.ID, cmd.Process, watchStop) })
 	}
 
 	// Eager-close stdin when adapter/decision-loop do not need the pipe.
@@ -228,7 +228,7 @@ func (e *ProcessExecutor) collectAndWaitOutput(
 	var wg sync.WaitGroup
 	outputLimiter := newRunOutputLimiter(e.maxRunOutputBytes)
 	wg.Add(1)
-	go e.publishOutput(&wg, run, outStore, outputLimiter, "stderr", proc.stderr)
+	safeGo("publishOutput.stderr", func() { e.publishOutput(&wg, run, outStore, outputLimiter, "stderr", proc.stderr) })
 
 	// Inject context budget for token tracking in stream parsers.
 	// Also inject RunProcessContext unconditionally — SDK adapters
@@ -238,11 +238,11 @@ func (e *ProcessExecutor) collectAndWaitOutput(
 
 	if proc.buildPlan.UseStructuredParser {
 		wg.Add(1)
-		go e.publishStructuredOutput(&wg, run, proc.stdout, proc.stdin, adapter, parserCtx, &parseErr)
+		safeGo("publishStructuredOutput", func() { e.publishStructuredOutput(&wg, run, proc.stdout, proc.stdin, adapter, parserCtx, &parseErr) })
 	} else {
 		// Raw capture: stdout goes to run.output.batch events
 		wg.Add(1)
-		go e.publishOutput(&wg, run, outStore, outputLimiter, "stdout", proc.stdout)
+		safeGo("publishOutput.stdout", func() { e.publishOutput(&wg, run, outStore, outputLimiter, "stdout", proc.stdout) })
 	}
 
 	// StdoutPipe/StderrPipe readers must finish before Wait closes the pipe
@@ -466,6 +466,6 @@ func (e *ProcessExecutor) handleFaultEscalation(
 	e.bus.Publish("run.fault_escalation.retry", runScope(*run),
 		faultEscalationRetryPayload(run.ID, newCount, e.faultEscalationCfg.MaxRetries))
 	slog.Warn("process: fault escalation auto-retry", "runId", run.ID, "retryCount", newCount, "maxRetries", e.faultEscalationCfg.MaxRetries)
-	go e.run(newCtx, *run, runCtx)
+	safeGo("run.faultEscalation", func() { e.run(newCtx, *run, runCtx) })
 	return true
 }

edge-server/internal/lifecycle/process_executor_cancel.go

@@ -47,7 +47,7 @@ func (e *ProcessExecutor) Cancel(runID string) CancelResult {
 		e.mu.Lock()
 		e.cancelDone[runID] = done
 		e.mu.Unlock()
-		go func() {
+		safeGo("cancelGrace", func() {
 			select {
 			case <-done:
 				return
@@ -68,7 +68,7 @@ func (e *ProcessExecutor) Cancel(runID string) CancelResult {
 			if _, err := proc.Wait(); planProcessWaitAfterKill(err).Log {
 				slog.Warn("process wait error after kill", "run_id", runID, "error", err)
 			}
-		}()
+		})
 	}
 
 	// Cancel the run context after the grace path is armed. This stops

edge-server/internal/lifecycle/process_executor_finish.go

@@ -58,6 +58,19 @@ func (e *ProcessExecutor) finish(runID string) {
 		delete(e.runOutputs, runID)
 	}
 	e.mu.Unlock()
+
+	// Clean up package-level hub callback state so a run that panicked and
+	// was recovered by safeGo (skipping fireHubDone/fireHubFail) does not
+	// leak hubCallbackQueues / hubStreamChunkSeq entries. The consumer
+	// goroutine's own defer also deletes; sync.Map.Delete is idempotent.
+	// Closing the queue channel unblocks a consumer stuck on range, letting
+	// it drain and exit cleanly.
+	if stateAny, ok := hubCallbackQueues.LoadAndDelete(runID); ok {
+		if state, ok := stateAny.(*hubCallbackQueueState); ok {
+			state.close()
+		}
+	}
+	hubStreamChunkSeq.Delete(runID)
 }
 
 // hasActiveChildren reports whether the given run has at least one registered

edge-server/internal/lifecycle/process_executor_hub_callback.go

@@ -249,6 +249,20 @@ func newHubCallbackQueueState(runID string) *hubCallbackQueueState {
 	}
 }
 
+// close idempotently closes the queue channel so the consumer goroutine (if
+// running) drains remaining jobs and exits. It is safe to call from finish()
+// when a panicked run skipped fireHubDone/fireHubFail and never enqueued a
+// terminal job — otherwise the queue entry and consumer would leak.
+func (s *hubCallbackQueueState) close() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed {
+		return
+	}
+	s.closed = true
+	close(s.ch)
+}
+
 // enqueueHubStreamJob appends a stream chunk to the run's FIFO without
 // blocking. Returns false when the queue is full (chunk dropped) or already
 // closed (terminal callback decided the run's delivery).

edge-server/internal/lifecycle/process_executor_start.go

@@ -20,7 +20,12 @@ func (e *ProcessExecutor) Start(run store.Run, runCtx RunProcessContext) error {
 	ctx, cancel := context.WithTimeout(context.Background(), e.runTimeout)
 	e.running[run.ID] = cancel
 
-	go e.run(ctx, run, bindRunProcessContext(runCtx, run))
+	// Spawn the run lifecycle goroutine through safeGo so a panic inside run()
+	// (adapter parse, emitter chain, output store) is recovered and logged
+	// instead of crashing the whole Edge process. run()'s deferred finish()
+	// still runs during panic unwinding before recover catches, so terminal
+	// state and package-level sync.Map hygiene are preserved.
+	safeGo("run", func() { e.run(ctx, run, bindRunProcessContext(runCtx, run)) })
 	return nil
 }
 

edge-server/internal/lifecycle/result_aggregator.go

@@ -106,12 +106,12 @@ func (ra *ResultAggregator) Start() (stop func()) {
 	ra.subID = subID
 
 	done := make(chan struct{})
-	go func() {
+	safeGo("resultAggregator", func() {
 		defer close(done)
 		for evt := range ch {
 			ra.handleEvent(evt)
 		}
-	}()
+	})
 
 	// Timeout fallback goroutine: periodically checks for parents whose
 	// children have exceeded the configured timeout. When found, emits
@@ -120,7 +120,7 @@ func (ra *ResultAggregator) Start() (stop func()) {
 	var timeoutDone chan struct{}
 	if ra.collector != nil {
 		timeoutDone = make(chan struct{})
-		go ra.runTimeoutCheck(timeoutDone)
+		safeGo("resultAggregatorTimeout", func() { ra.runTimeoutCheck(timeoutDone) })
 	}
 
 	return func() {