Cyrene_v2/internal/runtime/orchestration/spawntool_publish_log_test.go

package orchestration

import (
	"context"
	"errors"
	"log/slog"
	"sync"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
)

// 本文件锁住 M29 勘测 §7 的可观测性缺口（M32 顺手补掉的那一半）：`await` 的
// **每一条出口都必须留痕**。修复前「发布成功零日志、无总线静默 return」，于是
// 「后台结果到底发没发」只能靠负证据推断 —— 那正是这次缺陷最费时间的地方。
//
// 三条被测出口，与日志一一对应：
//   - 发布成功 ⇒ INFO `sub-agent results published`（带 source、父会话、条数、状态）；
//   - 没有总线 ⇒ WARN `... no bus`（以前是静默 return）；
//   - 发布失败 ⇒ WARN `publish sub-agent results`（带 err）。

// failingPublishBus 是一个 Publish 必定失败的 Bus：用于证明「发布失败不静默」。
type failingPublishBus struct {
	err error
}

func (b failingPublishBus) Publish(context.Context, string, any) error { return b.err }

func (b failingPublishBus) Subscribe(context.Context, string, bus.Handler) (func(), error) {
	return func() {}, nil
}

func (b failingPublishBus) Close() error { return nil }

// awaitLogSink 收集结构化日志记录（spawntool 的 logger 已带 source）。
type awaitLogSink struct {
	mu   sync.Mutex
	recs []slog.Record
}

// awaitLogCapture 是一个会把 WithAttrs 累积进记录里的处理器：source 等标签是经
// `logger.With` 挂上的，只读 Record.Attrs 会看不到它们。
type awaitLogCapture struct {
	sink  *awaitLogSink
	attrs []slog.Attr
}

func newAwaitLogCapture() *awaitLogCapture { return &awaitLogCapture{sink: &awaitLogSink{}} }

func (c *awaitLogCapture) Enabled(context.Context, slog.Level) bool { return true }

func (c *awaitLogCapture) Handle(_ context.Context, rec slog.Record) error {
	rec.AddAttrs(c.attrs...)
	c.sink.mu.Lock()
	defer c.sink.mu.Unlock()
	c.sink.recs = append(c.sink.recs, rec.Clone())
	return nil
}

func (c *awaitLogCapture) WithAttrs(as []slog.Attr) slog.Handler {
	merged := make([]slog.Attr, 0, len(c.attrs)+len(as))
	merged = append(merged, c.attrs...)
	merged = append(merged, as...)
	return &awaitLogCapture{sink: c.sink, attrs: merged}
}

func (c *awaitLogCapture) WithGroup(string) slog.Handler { return c }

// find 返回第一条级别匹配且 msg 等值的记录。
func (c *awaitLogCapture) find(level slog.Level, want string) (slog.Record, bool) {
	c.sink.mu.Lock()
	defer c.sink.mu.Unlock()
	for _, rec := range c.sink.recs {
		if rec.Level == level && rec.Message == want {
			return rec, true
		}
	}
	return slog.Record{}, false
}

// attr 读取一条记录上的字符串属性（含经 WithAttrs 挂上的标签）。
func (c *awaitLogCapture) attr(rec slog.Record, key string) (string, bool) {
	var out string
	var ok bool
	rec.Attrs(func(a slog.Attr) bool {
		if a.Key != key {
			return true
		}
		out, ok = a.Value.String(), true
		return false
	})
	return out, ok
}

// settledTool 构建一个派发并结算一个子代理的工具，返回工具、该子代理 id 与日志捕获。
func settledTool(t *testing.T, opts Options) (*SpawnTool, string, *awaitLogCapture) {
	t.Helper()
	s := New(opts)
	s.runner = &mockRunner{}
	sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "查一下 X", ParentSession: "web"})
	if err != nil {
		t.Fatalf("Spawn: %v", err)
	}
	capture := newAwaitLogCapture()
	return NewSpawnTool(s, WithSpawnLogger(slog.New(capture)), WithSpawnWait(time.Second)), sub.ID, capture
}

// TestAwaitLogsSuccessfulPublish 锁住「发成功要能从日志正面看出来」。
func TestAwaitLogsSuccessfulPublish(t *testing.T) {
	t.Parallel()
	b := bus.New()
	t.Cleanup(func() { _ = b.Close() })
	sp, id, capture := settledTool(t, Options{MaxParallel: 1, Bus: b})

	published := make(chan *SubagentResultEvent, 1)
	unsub, err := b.Subscribe(context.Background(), SubagentResultSubject,
		func(_ context.Context, e bus.Event) error {
			if ev, ok := e.Payload.(*SubagentResultEvent); ok {
				select {
				case published <- ev:
				default:
				}
			}
			return nil
		})
	if err != nil {
		t.Fatalf("subscribe: %v", err)
	}
	defer unsub()

	sp.await([]string{id}, "web")

	select {
	case ev := <-published:
		if ev.SessionID != "web" || ev.Status != "done" {
			t.Fatalf("payload = %+v, want the settled batch for session web", ev)
		}
	case <-time.After(waitTimeout):
		t.Fatal("the settled batch was never published")
	}
	rec, ok := capture.find(slog.LevelInfo, "sub-agent results published")
	if !ok {
		t.Fatal("a successful publish left no INFO: 'did it publish?' must be answerable from the log")
	}
	if src, _ := capture.attr(rec, "source"); src != "orchestration/spawn" {
		t.Fatalf("log source = %q, want %q (AGENTS §10: no bare logs)", src, "orchestration/spawn")
	}
	if session, _ := capture.attr(rec, "session"); session != "web" {
		t.Fatalf("log session = %q, want the parent session", session)
	}
	if status, _ := capture.attr(rec, "status"); status != "done" {
		t.Fatalf("log status = %q, want done", status)
	}
}

// TestAwaitWarnsWithoutBus 锁住「没有总线」不再是静默 return。
func TestAwaitWarnsWithoutBus(t *testing.T) {
	t.Parallel()
	sp, id, capture := settledTool(t, Options{MaxParallel: 1})
	if sp.sched.Bus() != nil {
		t.Skip("this test needs a scheduler without a bus")
	}
	sp.await([]string{id}, "web")
	rec, ok := capture.find(slog.LevelWarn, "sub-agent results not published: no bus")
	if !ok {
		t.Fatal("a scheduler with no bus dropped the results silently")
	}
	if session, _ := capture.attr(rec, "session"); session != "web" {
		t.Fatalf("log session = %q, want the parent session (traceability)", session)
	}
}

// TestAwaitWarnsOnPublishFailure 锁住「发布失败必须留下 err」。
func TestAwaitWarnsOnPublishFailure(t *testing.T) {
	t.Parallel()
	sp, id, capture := settledTool(t, Options{MaxParallel: 1, Bus: failingPublishBus{err: errors.New("bus down")}})
	sp.await([]string{id}, "web")

	rec, ok := capture.find(slog.LevelWarn, "publish sub-agent results")
	if !ok {
		t.Fatal("a failed publish left no WARN: the result would silently never arrive")
	}
	if msg, _ := capture.attr(rec, "error"); msg == "" {
		t.Fatal("the publish-failure WARN carries no error detail")
	}
}