Merge pull request 'Harden cross-instance HITL/cancel command paths and add disk-full rejection tests (U7)' (#164) from feature/u7-load-fault-release into main

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
+}

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 := ""

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 将 <dir>/artifacts 移至 <dir>/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 根 = <dataDir>/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)
+	}
+}

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) {