You've already forked agentic-coding-workflow
test(acp): supervisor integration with fake-acp-agent (happy/hang/crash)
This commit is contained in:
@@ -0,0 +1,245 @@
|
||||
//go:build integration
|
||||
|
||||
// supervisor_test.go 集成测试 supervisor 的 spawn / handshake / kill / monitor /
|
||||
// onExit 全链路。复用 cmd/fake-acp-agent 子进程：
|
||||
// - HappyPath：正常 spawn → handshake → 正常 Kill → 写 SessionExited
|
||||
// - HangAgent：FAKE_ACP_HANG=true 让 agent 永不响应 → handshake 超时 → Spawn 返错
|
||||
// - AgentCrashes：FAKE_ACP_CRASH_AFTER_PROMPTS=1 让 agent 处理一次 prompt 后 exit(1)
|
||||
//
|
||||
// 全部走 testcontainers PG，每测试一个独立容器（setupRepo 内 t.Cleanup）。
|
||||
package acp_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/yan1h/agent-coding-workflow/internal/acp"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/audit"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
||||
)
|
||||
|
||||
// newSupervisorForTest 构造一个绑定到 PG 真实 audit Recorder 的 Supervisor，
|
||||
// 使用静音 logger + 默认 SupervisorConfig（5s SpawnTimeout 给 fake agent 留 buffer）。
|
||||
// withAudit=false 时 Recorder 为 nil，对应 onExit 跳过 audit 路径。
|
||||
func newSupervisorForTest(t *testing.T, pool *pgxpool.Pool, repo acp.Repository, withAudit bool) *acp.Supervisor {
|
||||
t.Helper()
|
||||
var rec audit.Recorder
|
||||
if withAudit {
|
||||
rec = audit.NewPostgresRecorder(pool)
|
||||
}
|
||||
disp := notify.NewDispatcher(notifyRepoStub{}, slogDiscard())
|
||||
return acp.NewSupervisor(
|
||||
repo, rec, disp,
|
||||
nil, nil, // wtSvc / wsRepo: tests 用 IsMain=true，走 repo.ReleaseMainWorktree
|
||||
testEncryptor(t),
|
||||
acp.SupervisorConfig{
|
||||
SpawnTimeout: 5 * time.Second,
|
||||
KillGrace: 1 * time.Second,
|
||||
StderrBufferLines: 50,
|
||||
StderrTailBytes: 1000,
|
||||
EventMaxPayload: 65536,
|
||||
EventTruncateField: 4096,
|
||||
ShutdownGrace: 5 * time.Second,
|
||||
},
|
||||
slogDiscard(),
|
||||
)
|
||||
}
|
||||
|
||||
// TestSupervisor_HappyPath_Spawn_Handshake 验证：
|
||||
// 1. Spawn 启动 fake-acp-agent 并完成 initialize + session/new
|
||||
// 2. UpdateSessionRunning 把 agent_session_id="fake-session-id" + pid 写入 DB
|
||||
// 3. Kill 触发 monitorProcess + onExit，session 落到 SessionExited
|
||||
func TestSupervisor_HappyPath_Spawn_Handshake(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, pool := setupRepo(t)
|
||||
|
||||
uid := mustInsertUser(t, ctx, pool, "sup1@local", false)
|
||||
pid, wsid := mustInsertProjectWorkspace(t, ctx, pool, uid)
|
||||
|
||||
bin := fakeAgentBinary(t)
|
||||
k, err := repo.CreateAgentKind(ctx, &acp.AgentKind{
|
||||
ID: uuid.New(), Name: "fake1", DisplayName: "Fake",
|
||||
BinaryPath: bin, Args: []string{},
|
||||
Enabled: true, CreatedBy: uid,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
cwd := t.TempDir()
|
||||
sess, err := repo.InsertSession(ctx, &acp.Session{
|
||||
ID: uuid.New(), WorkspaceID: wsid, ProjectID: pid,
|
||||
AgentKindID: k.ID, UserID: uid,
|
||||
Branch: "main", CwdPath: cwd, IsMainWorktree: true, Status: acp.SessionStarting,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// onExit 会调 ReleaseMainWorktree → 必须先 Acquire 才能 Release（CAS 语义）。
|
||||
ok, err := repo.AcquireMainWorktree(ctx, sess.ID, wsid)
|
||||
require.NoError(t, err)
|
||||
require.True(t, ok)
|
||||
|
||||
sup := newSupervisorForTest(t, pool, repo, true)
|
||||
proc, err := sup.Spawn(ctx, sess, k)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, proc)
|
||||
|
||||
got, err := repo.GetSessionByID(ctx, sess.ID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, acp.SessionRunning, got.Status)
|
||||
require.NotNil(t, got.AgentSessionID)
|
||||
assert.Equal(t, "fake-session-id", *got.AgentSessionID)
|
||||
require.NotNil(t, got.PID)
|
||||
assert.NotZero(t, *got.PID)
|
||||
|
||||
sup.Kill(ctx, sess.ID, 1*time.Second)
|
||||
|
||||
// monitorProcess + onExit 在 goroutine 里跑；轮询直到 DB 反映终态。
|
||||
deadline := time.After(5 * time.Second)
|
||||
for {
|
||||
got, err = repo.GetSessionByID(ctx, sess.ID)
|
||||
require.NoError(t, err)
|
||||
if got.Status == acp.SessionExited || got.Status == acp.SessionCrashed {
|
||||
break
|
||||
}
|
||||
select {
|
||||
case <-deadline:
|
||||
t.Fatalf("session never reached terminal state, last status=%s", got.Status)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
assert.Equal(t, acp.SessionExited, got.Status)
|
||||
}
|
||||
|
||||
// TestSupervisor_HangAgent_HandshakeTimeout 验证：FAKE_ACP_HANG agent 永远不响应；
|
||||
// Spawn 在 SpawnTimeout 内返回 error，Kill 在内部触发 monitor → onExit → DB 终态。
|
||||
func TestSupervisor_HangAgent_HandshakeTimeout(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, pool := setupRepo(t)
|
||||
|
||||
uid := mustInsertUser(t, ctx, pool, "sup2@local", false)
|
||||
pid, wsid := mustInsertProjectWorkspace(t, ctx, pool, uid)
|
||||
|
||||
bin := fakeAgentBinary(t)
|
||||
k, err := repo.CreateAgentKind(ctx, &acp.AgentKind{
|
||||
ID: uuid.New(), Name: "fake-hang", DisplayName: "Fake",
|
||||
BinaryPath: bin, Args: []string{},
|
||||
EncryptedEnv: encryptForTest(t, map[string]string{"FAKE_ACP_HANG": "true"}),
|
||||
Enabled: true, CreatedBy: uid,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
sess, err := repo.InsertSession(ctx, &acp.Session{
|
||||
ID: uuid.New(), WorkspaceID: wsid, ProjectID: pid,
|
||||
AgentKindID: k.ID, UserID: uid,
|
||||
Branch: "main", CwdPath: t.TempDir(), IsMainWorktree: true, Status: acp.SessionStarting,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
ok, err := repo.AcquireMainWorktree(ctx, sess.ID, wsid)
|
||||
require.NoError(t, err)
|
||||
require.True(t, ok)
|
||||
|
||||
// 用更短的 SpawnTimeout 加速测试（默认 5s 够，这里 1s 避免 CI 等太久）。
|
||||
disp := notify.NewDispatcher(notifyRepoStub{}, slogDiscard())
|
||||
sup := acp.NewSupervisor(
|
||||
repo, audit.NewPostgresRecorder(pool), disp,
|
||||
nil, nil, testEncryptor(t),
|
||||
acp.SupervisorConfig{
|
||||
SpawnTimeout: 1 * time.Second,
|
||||
KillGrace: 500 * time.Millisecond,
|
||||
StderrBufferLines: 10,
|
||||
StderrTailBytes: 500,
|
||||
EventMaxPayload: 65536,
|
||||
EventTruncateField: 4096,
|
||||
ShutdownGrace: 2 * time.Second,
|
||||
},
|
||||
slogDiscard(),
|
||||
)
|
||||
|
||||
_, err = sup.Spawn(ctx, sess, k)
|
||||
require.Error(t, err, "hang agent must trigger handshake timeout")
|
||||
|
||||
deadline := time.After(5 * time.Second)
|
||||
for {
|
||||
got, gerr := repo.GetSessionByID(ctx, sess.ID)
|
||||
require.NoError(t, gerr)
|
||||
if got.Status == acp.SessionCrashed || got.Status == acp.SessionExited {
|
||||
// SIGKILL on hang agent 不属于 \"agent crashed by itself\"，
|
||||
// supervisor 会标 KilledByUs → exited。两者都接受。
|
||||
assert.Contains(t, []acp.SessionStatus{acp.SessionCrashed, acp.SessionExited}, got.Status)
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-deadline:
|
||||
t.Fatalf("session never reached terminal state, last status=%s", got.Status)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestSupervisor_AgentCrashes 验证：fake agent 在处理 1 次 prompt 后 os.Exit(1)；
|
||||
// monitorProcess 检测到非主动 Kill → status=SessionCrashed + exit_code=1，
|
||||
// onExit 写 DB + 调 Dispatcher.Dispatch + audit.Record。
|
||||
func TestSupervisor_AgentCrashes(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
repo, pool := setupRepo(t)
|
||||
|
||||
uid := mustInsertUser(t, ctx, pool, "sup3@local", false)
|
||||
pid, wsid := mustInsertProjectWorkspace(t, ctx, pool, uid)
|
||||
|
||||
bin := fakeAgentBinary(t)
|
||||
k, err := repo.CreateAgentKind(ctx, &acp.AgentKind{
|
||||
ID: uuid.New(), Name: "fake-crash", DisplayName: "Fake",
|
||||
BinaryPath: bin, Args: []string{},
|
||||
EncryptedEnv: encryptForTest(t, map[string]string{"FAKE_ACP_CRASH_AFTER_PROMPTS": "1"}),
|
||||
Enabled: true, CreatedBy: uid,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
sess, err := repo.InsertSession(ctx, &acp.Session{
|
||||
ID: uuid.New(), WorkspaceID: wsid, ProjectID: pid,
|
||||
AgentKindID: k.ID, UserID: uid,
|
||||
Branch: "main", CwdPath: t.TempDir(), IsMainWorktree: true, Status: acp.SessionStarting,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
ok, err := repo.AcquireMainWorktree(ctx, sess.ID, wsid)
|
||||
require.NoError(t, err)
|
||||
require.True(t, ok)
|
||||
|
||||
sup := newSupervisorForTest(t, pool, repo, true)
|
||||
proc, err := sup.Spawn(ctx, sess, k)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, proc)
|
||||
|
||||
// 通过 relay.SendClient 发一条 prompt → fake agent 处理后 exit(1)。
|
||||
idJSON, _ := json.Marshal("client-1")
|
||||
err = proc.Relay.SendClient(&acp.Message{
|
||||
JSONRPC: "2.0", ID: idJSON, Method: "session/prompt",
|
||||
Params: json.RawMessage(`{"sessionId":"fake-session-id","prompt":[{"type":"text","text":"x"}]}`),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
deadline := time.After(10 * time.Second)
|
||||
for {
|
||||
got, gerr := repo.GetSessionByID(ctx, sess.ID)
|
||||
require.NoError(t, gerr)
|
||||
if got.Status == acp.SessionCrashed {
|
||||
require.NotNil(t, got.ExitCode)
|
||||
assert.Equal(t, int32(1), *got.ExitCode)
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-deadline:
|
||||
t.Fatalf("crash not detected, last status=%s", got.Status)
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
//go:build integration
|
||||
|
||||
// test_helpers_test.go 是 supervisor_test.go 的共享脚手架：
|
||||
// - notifyRepoStub：满足 notify.Repository（5 方法），让 dispatcher 落库变成空操作
|
||||
// - slogDiscard：丢弃日志的 logger
|
||||
// - encryptForTest：用与 AgentKindService 相同格式产出加密 env（map → JSON → AES-GCM）
|
||||
// - fakeAgentBinary：构建 cmd/fake-acp-agent 二进制供子进程 spawn
|
||||
//
|
||||
// 集成测试专用，不参与 go test 默认 build。
|
||||
package acp_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
||||
)
|
||||
|
||||
// notifyRepoStub 实现 notify.Repository 全部 5 方法。supervisor.onExit
|
||||
// 会通过 dispatcher 落库一条 crashed 通知，落库走 stub → 永远成功 + 不存。
|
||||
type notifyRepoStub struct{}
|
||||
|
||||
func (notifyRepoStub) Insert(context.Context, notify.Message) error { return nil }
|
||||
func (notifyRepoStub) List(context.Context, uuid.UUID, bool, int) ([]notify.Message, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (notifyRepoStub) CountUnread(context.Context, uuid.UUID) (int, error) { return 0, nil }
|
||||
func (notifyRepoStub) MarkRead(context.Context, uuid.UUID, uuid.UUID) error { return nil }
|
||||
func (notifyRepoStub) MarkAllRead(context.Context, uuid.UUID) error { return nil }
|
||||
|
||||
// slogDiscard 返回静音 logger，避免测试输出污染。
|
||||
func slogDiscard() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }
|
||||
|
||||
// encryptForTest 以 AgentKindService 同格式（JSON marshal → AES-GCM）产出
|
||||
// EncryptedEnv 字节，让 supervisor.decryptEnv 还原为同一 map。
|
||||
func encryptForTest(t *testing.T, m map[string]string) []byte {
|
||||
t.Helper()
|
||||
enc := testEncryptor(t)
|
||||
plain, err := json.Marshal(m)
|
||||
require.NoError(t, err)
|
||||
out, err := enc.Encrypt(plain)
|
||||
require.NoError(t, err)
|
||||
return out
|
||||
}
|
||||
|
||||
// repoRoot 从测试 cwd 向上找 go.mod，返回仓库根目录。t.TempDir + go build 用。
|
||||
func repoRoot(t *testing.T) string {
|
||||
t.Helper()
|
||||
wd, err := os.Getwd()
|
||||
require.NoError(t, err)
|
||||
for {
|
||||
if _, err := os.Stat(filepath.Join(wd, "go.mod")); err == nil {
|
||||
return wd
|
||||
}
|
||||
parent := filepath.Dir(wd)
|
||||
if parent == wd {
|
||||
t.Fatal("go.mod not found upward")
|
||||
}
|
||||
wd = parent
|
||||
}
|
||||
}
|
||||
|
||||
// fakeAgentBinary 编译 cmd/fake-acp-agent 到 t.TempDir() 并返回路径。
|
||||
// 测试结束后 t.TempDir 自动清理。
|
||||
func fakeAgentBinary(t *testing.T) string {
|
||||
t.Helper()
|
||||
bin := filepath.Join(t.TempDir(), "fake-acp-agent")
|
||||
if runtime.GOOS == "windows" {
|
||||
bin += ".exe"
|
||||
}
|
||||
cmd := exec.Command("go", "build", "-o", bin, "./cmd/fake-acp-agent")
|
||||
cmd.Dir = repoRoot(t)
|
||||
out, err := cmd.CombinedOutput()
|
||||
require.NoError(t, err, "build fake-acp-agent: %s", out)
|
||||
return bin
|
||||
}
|
||||
Reference in New Issue
Block a user