diff --git a/internal/runmgr/multi_instance_test.go b/internal/runmgr/multi_instance_test.go index 5a9b5a1..51fd3a9 100644 --- a/internal/runmgr/multi_instance_test.go +++ b/internal/runmgr/multi_instance_test.go @@ -174,3 +174,40 @@ func TestCommandOnlyConsumedByOwner(t *testing.T) { } _ = context.Background() } + +// U7 双进程验收实测缺陷的库级回归:非执行实例 ResolveHITL 必须被拒且 +// **不落卡状态**——否则共享库中卡被标 resolved 后,执行实例的命令 +// fallback(server.resolveHITL 的 !card.Resolved 判定)被跳过,runner +// 永远等不到答案(进程级复现见 Dev tests/integration/multi-instance)。 +func TestCrossInstanceResolveDoesNotSettleCard(t *testing.T) { + st := pgSchemaStore(t) + a := newOwnedManager(t, st, "instA") + b := newOwnedManager(t, st, "instB") + conv := recoveryConversation(t, st) + run := recoveryRun(t, st, conv, core.RunWaitingHITL) + card := core.HITLCard{ + RunID: run.ID, ConversationID: conv.ID, Question: "跨实例审批", + Kind: core.HITLKindChoice, + Choices: []core.HITLChoice{ + {ID: "md", Label: "Markdown"}, + }, + CreatedAt: time.Now().UTC().Format(time.RFC3339), + } + card.ID = store.NewID("hitl") + if err := st.SaveHITLCard(store.OwnerDefault, &card); err != nil { + t.Fatal(err) + } + + // B(非执行实例):明确拒绝,卡保持未决,留给命令总线转交 A。 + if _, _, err := b.ResolveHITL(store.OwnerDefault, string(card.ID), core.HITLResponse{ChoiceID: "md"}); err == nil || !strings.Contains(err.Error(), "not executing locally") { + t.Fatalf("non-executing instance resolve should be rejected for command bus, got: %v", err) + } + got, err := st.GetHITLCard(store.OwnerDefault, string(card.ID)) + if err != nil { + t.Fatal(err) + } + if got.Resolved { + t.Fatal("non-executing instance resolve settled the card (would strand the executing runner)") + } + _ = a +} diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index ce0b423..fcdd23c 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -1561,6 +1561,16 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv if err != nil { return "", false, err } + // U7 双进程验收实测缺陷修复:run 不在本进程执行时不得先落 resolve—— + // 卡一旦标记 resolved,另一实例 server 层的命令 fallback(resolveHITL 的 + // !card.Resolved 判定)即被跳过,执行实例的 runner 永远等不到答案, + // run 只能等 HITL 超时。此处提前返回未决错误,让命令总线转交执行实例。 + m.mu.Lock() + _, local := m.active[run.ID] + m.mu.Unlock() + if !local { + return "", false, fmt.Errorf("runmgr: run %s not executing locally (cross-instance resolve goes through command bus)", run.ID) + } kind := "choice" answer := "" diff --git a/internal/server/disk_full_test.go b/internal/server/disk_full_test.go new file mode 100644 index 0000000..e1e8b00 --- /dev/null +++ b/internal/server/disk_full_test.go @@ -0,0 +1,217 @@ +package server + +import ( + "bytes" + "io" + "mime/multipart" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "agent-system-backend/internal/bus" + "agent-system-backend/internal/cmdbus" + "agent-system-backend/internal/plugins" + "agent-system-backend/internal/profiles" + "agent-system-backend/internal/runmgr" + "agent-system-backend/internal/runtime" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + "agent-system-backend/internal/tools" +) + +// 协作存储 U7 故障矩阵(§7「磁盘满/备份恢复/整机失效」行的磁盘满维度): +// 对象目录不可写时,写入类入口必须**清楚拒绝**(明确 5xx,绝不 2xx 静默或 +// 半成品落库),且恢复后不留后遗症。注入不依赖 chmod(CI 容器以 root 运行, +// 0500 目录仍可写):把 artifacts 目录移走并在原位放置同名普通文件, +// blobstore 临时文件写入得到 ENOTDIR——与磁盘满同为 os.WriteFile 层的 +// 持久性失败。 +// +// 覆盖: +// 1. 上传(POST /api/projects/{id}/uploads)在对象目录失效时 5xx, +// artifacts 表零新增、对象目录零残留;恢复后同一上传成功。 +// 2. 工件内容覆写(PUT /api/artifacts/{id}/content)同样拒绝, +// 且原内容保持可读(拒绝不产生半写状态)。 + +// breakArtifactsDir 将 /artifacts 移至 /artifacts.moved 并在原位 +// 放置同名文件;返回恢复函数。 +func breakArtifactsDir(t *testing.T, dataDir string) (restore func()) { + t.Helper() + dir := filepath.Join(dataDir, "artifacts") + moved := dir + ".moved" + if err := os.Rename(dir, moved); err != nil { + t.Fatalf("move artifacts dir: %v", err) + } + if err := os.WriteFile(dir, []byte("not a directory"), 0o644); err != nil { + t.Fatalf("plant file at artifacts path: %v", err) + } + return func() { + if err := os.Remove(dir); err != nil { + t.Fatalf("remove planted file: %v", err) + } + if err := os.Rename(moved, dir); err != nil { + t.Fatalf("restore artifacts dir: %v", err) + } + } +} + +// diskFullServer 自建可控 dataDir 的服务(artifacts 根 = /artifacts, +// 见 store.OpenSQLite 的 finishOpen 约定),其余装配与 admission/e2e 一致。 +func diskFullServer(t *testing.T) (http.Handler, *store.Store, string) { + t.Helper() + dir := t.TempDir() + st, err := store.OpenSQLite(filepath.Join(dir, "test.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + reg := tools.NewRegistry() + reg.RegisterBuiltin() + setSvc := settings.NewService(st) + profSvc := profiles.New(st) + profSvc.EnsureDefault() + mgr := runmgr.New(st, bus.New(), reg, profSvc, setSvc, &runtime.MockRunner{}, &runtime.MockRunner{}, &runtime.MockRunner{}, &runtime.MockSubagent{}) + t.Cleanup(mgr.Shutdown) + ph := plugins.New(st, reg, setSvc, filepath.Join(dir, "plugins")) + srv := New(st, mgr, reg, profSvc, setSvc, ph) + srv.Bus = cmdbus.New(st, nil, nil) + srv.WireBuiltinCommands() + return srv.Handler(), st, dir +} + +func multipartUpload(t *testing.T, h http.Handler, projectID, filename string) *httptest.ResponseRecorder { + t.Helper() + var buf bytes.Buffer + mw := multipart.NewWriter(&buf) + fw, err := mw.CreateFormFile("file", filename) + if err != nil { + t.Fatal(err) + } + if _, err := io.Copy(fw, bytes.NewReader([]byte("disk-full probe content: "+filename))); err != nil { + t.Fatal(err) + } + mw.Close() + req := httptest.NewRequest(http.MethodPost, "/api/projects/"+projectID+"/uploads", &buf) + req.Header.Set("Content-Type", mw.FormDataContentType()) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +func artifactCount(t *testing.T, st *store.Store, projectID string) (int, map[string]bool) { + t.Helper() + arts, err := st.ListArtifacts(store.OwnerDefault, projectID) + if err != nil { + t.Fatal(err) + } + names := map[string]bool{} + for _, a := range arts { + names[a.Name] = true + } + return len(arts), names +} + +func objectsIn(t *testing.T, dir string) map[string]bool { + t.Helper() + out := map[string]bool{} + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + for _, e := range entries { + out[e.Name()] = true + } + return out +} + +func TestUploadRejectsCleanlyWhenObjectDirUnwritable(t *testing.T) { + h, st, dir := diskFullServer(t) + code, out := doJSON(t, h, "POST", "/api/projects", map[string]any{"name": "disk-full"}) + if code != http.StatusOK { + t.Fatalf("create project: %d %v", code, out) + } + projID := out["id"].(string) + + seed := multipartUpload(t, h, projID, "seed.txt") + if seed.Code != http.StatusOK { + t.Fatalf("seed upload: %d %.200s", seed.Code, seed.Body.String()) + } + beforeCount, _ := artifactCount(t, st, projID) + beforeObjs := objectsIn(t, filepath.Join(dir, "artifacts")) + + restore := breakArtifactsDir(t, dir) + rec := multipartUpload(t, h, projID, "probe-diskfull.txt") + if rec.Code < 500 { + t.Fatalf("upload under broken object dir: status %d (want explicit 5xx rejection), body %.200s", rec.Code, rec.Body.String()) + } + if got, _ := artifactCount(t, st, projID); got != beforeCount { + t.Fatalf("rejected upload left DB rows: %d -> %d", beforeCount, got) + } + if got := objectsIn(t, filepath.Join(dir, "artifacts.moved")); len(got) != len(beforeObjs) { + t.Fatalf("rejected upload left object residue: %v -> %v", beforeObjs, got) + } + restore() + + after := multipartUpload(t, h, projID, "probe-diskfull-recovered.txt") + if after.Code != http.StatusOK { + t.Fatalf("upload after recovery: %d %.200s", after.Code, after.Body.String()) + } + if got, names := artifactCount(t, st, projID); got != beforeCount+1 || !names["probe-diskfull-recovered.txt"] { + t.Fatalf("recovered upload missing row: count %d names %v", got, names) + } +} + +func TestArtifactOverwriteRejectsCleanlyWhenObjectDirUnwritable(t *testing.T) { + h, st, dir := diskFullServer(t) + code, out := doJSON(t, h, "POST", "/api/projects", map[string]any{"name": "disk-full-2"}) + if code != http.StatusOK { + t.Fatalf("create project: %d %v", code, out) + } + projID := out["id"].(string) + + seed := multipartUpload(t, h, projID, "base.txt") + if seed.Code != http.StatusOK { + t.Fatalf("seed upload: %d %.200s", seed.Code, seed.Body.String()) + } + arts, err := st.ListArtifacts(store.OwnerDefault, projID) + if err != nil || len(arts) != 1 { + t.Fatalf("seed artifacts: %d err %v", len(arts), err) + } + artID := string(arts[0].ID) + + getContent := func() string { + req := httptest.NewRequest(http.MethodGet, "/api/artifacts/"+artID+"/content", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("get content: %d %.200s", rec.Code, rec.Body.String()) + } + return rec.Body.String() + } + before := getContent() + + restore := breakArtifactsDir(t, dir) + put := httptest.NewRequest(http.MethodPut, "/api/artifacts/"+artID+"/content", bytes.NewReader([]byte("half-written?"))) + putRec := httptest.NewRecorder() + h.ServeHTTP(putRec, put) + if putRec.Code < 500 { + t.Fatalf("overwrite under broken object dir: status %d (want explicit 5xx)", putRec.Code) + } + restore() + + // 原位文件占位已撤;拒绝期间不允许半写污染读取路径。 + if got := getContent(); got != before { + t.Fatalf("rejected overwrite altered readable content: %q -> %q", before, got) + } + + put2 := httptest.NewRequest(http.MethodPut, "/api/artifacts/"+artID+"/content", bytes.NewReader([]byte("recovered write"))) + put2Rec := httptest.NewRecorder() + h.ServeHTTP(put2Rec, put2) + if put2Rec.Code != http.StatusOK { + t.Fatalf("overwrite after recovery: %d %.200s", put2Rec.Code, put2Rec.Body.String()) + } + if got := getContent(); got != "recovered write" { + t.Fatalf("content after recovered overwrite: %q", got) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index dd7acd2..5f38696 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1026,23 +1026,26 @@ func (s *Server) listAudits(w http.ResponseWriter, r *http.Request) { func (s *Server) cancelRun(w http.ResponseWriter, r *http.Request) { runID := r.PathValue("id") - err := s.mgr.CancelRun(runID) - if err == nil { - writeOKOr(w, nil) - return - } - // 跨实例取消(协作存储 U5/KTD5):run 不在本实例执行(无句柄)且 PG 档 - // ——写持久命令,由执行实例消费;API 立即受理(命令可见、唯一消费)。 - if s.st.Dialect() == "postgres" && s.mgr.RunActiveLocally(runID) == false { - if cerr := s.st.EnqueueCommand(store.NewID("cmd"), "cancel", runID, "", 0, + // 跨实例取消(协作存储 U5/KTD5)**前置**:run 不在本实例执行(无句柄) + // 且 PG 档——写持久命令,由执行实例消费;API 立即受理(202)。 + // 不得先调本地 CancelRun:非执行实例无句柄时它返回 nil 的「假成功」 + // 会吞掉命令路径,run 永远停不下(U7 双进程验收实测)。 + // conversation_id 必须回填真实值:消费侧按执行权 JOIN 匹配,空串永远 + // 无法送达(同批实测)。 + if s.st.Dialect() == "postgres" && !s.mgr.RunActiveLocally(runID) { + convID := "" + if run, gerr := s.st.GetRun(s.owner(r), runID); gerr == nil { + convID = string(run.ConversationID) + } + if cerr := s.st.EnqueueCommand(store.NewID("cmd"), "cancel", runID, convID, 0, runmgr.CancelCommandPayload{Owner: s.owner(r), RunID: runID}); cerr != nil { - writeStoreErr(w, err) + writeStoreErr(w, cerr) return } writeJSON(w, http.StatusAccepted, map[string]any{"accepted": true, "via": "command"}) return } - writeStoreErr(w, err) + writeOKOr(w, s.mgr.CancelRun(runID)) } func (s *Server) getRun(w http.ResponseWriter, r *http.Request) {