fix(engine): up --wait 只按本次运行确认就绪

docs/cli.md

@@ -480,7 +480,13 @@ this time that time out enter stop cleanup; existing runs are not stopped or res
 wait failed. The cleanup names each run it started by `run_id`: a run another controller started for
 the same service while the wait was still running is a different run and is left alone. A daemon that
 does not announce `groups.kill.expect_runs` cannot scope a stop to named runs, so it is not sent a
-name-only cleanup: the command fails and says the timed-out runs are still running. The `started` /
+name-only cleanup: the command fails and says the timed-out runs are still running. Readiness is
+attributed by the same identity: when the service row's live run is not the run this wait is about, the
+wait ends immediately with state `failed` and reason `run_replaced`, because the replacement's listener
+and health are its own evidence, not the original run's; nothing is stopped then, since the replacement
+is not this invocation's run. A row that names no run cannot confirm a named one: a listener not
+attributed to the run keeps the wait going instead of counting as its readiness. A chunk with no
+`run_id` at all carries no identity to compare, so the service row is read as before. The `started` /
 `skipped` summary describes the start action, and readiness failures appear separately in `errors`
 and in each service's `state`. For runs newly started this time that
 were cleaned up, the engine writes the same `ready_timeout` into the service's last exit facts, so

docs/cli.zh.md

@@ -400,7 +400,11 @@ cwd 中执行，按 `cmd:` 的规则拆成 argv（不经过 shell），支持 `$
 已经存在的运行不会因这次等待失败而被停止或重启。清理按 `run_id` 只停止本次启动的那个运行：
 等待期间另一个控制者为同一服务启动的新运行是另一个运行，不会被停掉。daemon 若未声明
 `groups.kill.expect_runs`，就无法把停止限定到指定运行，因此不会收到只带服务名的清理：
-命令失败，并说明超时的运行仍在运行。`started` / `skipped` 汇总描述启动动作，
+命令失败，并说明超时的运行仍在运行。就绪判定同样按运行身份归属：状态行里的运行不是这次等待所指的
+运行时就立即以状态 `failed`、原因 `run_replaced` 结束，因为替换运行的监听与健康是它自己的证据，
+不能算作原运行的；此时不停止任何运行，替换运行不是本次调用启动的。状态行没有指明运行时也无法确认具名运行：
+无法归属到该运行的监听只会让等待继续，不会被算作它的就绪。完全没有 `run_id` 的 chunk
+没有可比对的身份，仍按服务行判断。`started` / `skipped` 汇总描述启动动作，
 就绪判定失败另外出现在 `errors` 与逐服务 `state` 中。对本次新启动且被清理的运行，
 引擎把同一个 `ready_timeout` 写入服务的最后退出事实，因此 `status`、桌面服务行、attention 和事件流
 不会把它显示成普通的手动停止。健康事实带有 `status`、`code`、`reason` 与 `observed_at`，

engine/internal/cmd/up_ready.go

@@ -14,6 +14,11 @@ import (
 	"github.com/sheathedsharp/option-berth/engine/internal/state"
 )
 
+// startRows reads the state a readiness wait decides on. Production reads the
+// same snapshot status exposes; a test replaces it so a readiness rule can be
+// checked without a daemon or the host's listener table.
+var startRows = groupRowsObserved
+
 // startTimeout is one service this wait gave up on: the name it was started
 // under and the run this invocation started. The run id is what authorizes the
 // cleanup — a service name alone would also match a run another controller
@@ -62,7 +67,7 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g
 	fallbackSince := waitStart.UTC().Truncate(time.Second)
 	var timedOut []startTimeout
 	for {
-		pp, gg, obs, err := groupRowsObserved(ctx)
+		pp, gg, obs, err := startRows(ctx)
 		if obs != nil && obs.State != string(ports.ObservationObservable) {
 			// Waiting the full timeout would end by stopping the run it just
 			// started. "Cannot see" is a different outcome from "not ready".
@@ -96,7 +101,17 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g
 				if group == nil || svc == nil {
 					continue
 				}
-				if ready, _ := serviceReady(*group, portsNow, name); ready {
+				// Readiness is the selected run's, not the service name's. A
+				// row that names another run is a replacement, and a row that
+				// names none cannot confirm a named one: its listener and its
+				// health answer for whatever it does name.
+				if replacement, ok := replacementRun(selection, svc); ok {
+					markStartChunkFailed(chunks, summary, name, "run_replaced",
+						startReplacedMessage(selection, replacement), nil)
+					delete(selected, name)
+					continue
+				}
+				if ready, _ := serviceReady(*group, portsNow, name); ready && startRowConfirms(selection, svc) {
 					markStartChunkReady(chunks, name)
 					delete(selected, name)
 					continue
@@ -198,6 +213,41 @@ func startTimeoutMessage(selection startSelection, name string, timeout time.Dur
 		selection.budget, name)
 }
 
+// replacementRun names the live run that took this selection's place. A row
+// for the same service that names another run answers for that run: its
+// listener and its health are evidence about the replacement, never about the
+// run this wait is about. Only two known run ids can disagree — a missing id
+// on either side is no identity evidence, and the wait keeps reading the row.
+func replacementRun(selection startSelection, svc *state.Service) (string, bool) {
+	if selection.runID == "" || svc == nil || !svc.Running || svc.RunID == nil {
+		return "", false
+	}
+	if *svc.RunID == "" || *svc.RunID == selection.runID {
+		return "", false
+	}
+	return *svc.RunID, true
+}
+
+// startRowConfirms reports whether the service row is evidence about the
+// selected run. A row that names no run cannot confirm a named one: whatever
+// is listening there is some process on the service's address, not this run's
+// readiness. A selection with no run id has nothing to compare and keeps
+// reading the row.
+func startRowConfirms(selection startSelection, svc *state.Service) bool {
+	if selection.runID == "" {
+		return true
+	}
+	return svc != nil && svc.RunID != nil && *svc.RunID == selection.runID
+}
+
+// startReplacedMessage states what ended the wait. It does not call the
+// replacement unhealthy: the point is that it is not the run this wait was
+// asked to confirm, so it cannot answer for it.
+func startReplacedMessage(selection startSelection, replacement string) string {
+	return fmt.Sprintf("the run this wait is about (%s) was replaced by %s while waiting",
+		selection.runID, replacement)
+}
+
 // startChunkReadyTimeout is the readiness budget the daemon published for one
 // service. A daemon predating ready_timeout omits it; the client then falls
 // back to the same default the manifest uses.

engine/internal/cmd/up_ready_test.go

@@ -0,0 +1,181 @@
+package cmd
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/sheathedsharp/option-berth/engine/internal/daemon/rpc"
+	"github.com/sheathedsharp/option-berth/engine/internal/groups"
+	"github.com/sheathedsharp/option-berth/engine/internal/ports"
+	"github.com/sheathedsharp/option-berth/engine/internal/state"
+)
+
+// startRowsFrom replaces the wait's observation with one group row set, so a
+// readiness rule is decided on rows a test controls instead of a daemon or the
+// host's listener table.
+func startRowsFrom(t *testing.T, groups ...state.Group) {
+	t.Helper()
+	prev := startRows
+	startRows = func(context.Context) ([]ports.ListeningPort, []state.Group, *rpc.ObservationStatus, error) {
+		return nil, groups, &rpc.ObservationStatus{State: string(ports.ObservationObservable)}, nil
+	}
+	t.Cleanup(func() { startRows = prev })
+}
+
+// startManifest writes one project manifest and returns its path.
+func startManifest(t *testing.T, services string) string {
+	t.Helper()
+	dir := t.TempDir()
+	path := filepath.Join(dir, groups.ConfigName)
+	if err := os.WriteFile(path, []byte("name: demo\nservices:\n"+services), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}
+
+// TestWaitForStartReadyCreditsOnlyTheRunItIsAbout: the wait is about a run, not
+// about whatever answers to the service's name. R1 was started here and is
+// still waiting; another controller restarts the same service as R2, and R2
+// becomes healthy. R2's health is evidence about R2 — reading it as R1's
+// readiness reports a service that was never confirmed, and the run that is
+// actually healthy is not this invocation's to stop.
+func TestWaitForStartReadyCreditsOnlyTheRunItIsAbout(t *testing.T) {
+	port, healthz := 19001, "/healthz"
+	for _, tc := range []struct {
+		name    string
+		chunk   rpc.GroupsStartChunk
+		service state.Service
+	}{
+		{
+			name:  "a healthy replacement listener",
+			chunk: rpc.GroupsStartChunk{Service: "api", State: "started", RunID: "run-r1"},
+			service: state.Service{PortActual: &port, Health: &healthz,
+				HealthStatus: &state.Health{Status: state.HealthOK}},
+		},
+		{
+			name:    "a healthy replacement worker",
+			chunk:   rpc.GroupsStartChunk{Service: "api", State: "started", RunID: "run-r1"},
+			service: state.Service{},
+		},
+		{
+			name:  "a skipped run the chunk named",
+			chunk: rpc.GroupsStartChunk{Service: "api", State: "skipped", Skipped: true, RunID: "run-r1"},
+			service: state.Service{PortActual: &port, Health: &healthz,
+				HealthStatus: &state.Health{Status: state.HealthOK}},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			path := startManifest(t, "  - name: api\n    cmd: echo api\n")
+			replacement := "run-r2"
+			svc := tc.service
+			svc.Name = "api"
+			svc.Running = true
+			svc.RunID = &replacement
+			startRowsFrom(t, state.Group{Name: "demo", ConfigPath: &path, Services: []state.Service{svc}})
+
+			chunk := tc.chunk
+			chunk.ReadyTimeout = "250ms"
+			chunks := []rpc.GroupsStartChunk{chunk}
+			summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}}
+			timedOut, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path},
+				nil, chunks, &summary, 0, false)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if chunks[0].Ready != nil && *chunks[0].Ready {
+				t.Fatalf("run %s was reported ready on run %s's evidence", "run-r1", replacement)
+			}
+			if chunks[0].State != "failed" || chunks[0].Reason != "run_replaced" {
+				t.Fatalf("chunk = %+v, want a run_replaced failure naming the replacement", chunks[0])
+			}
+			if len(summary.Errors) != 1 || summary.Errors[0] != "api" {
+				t.Fatalf("summary errors = %v, want api", summary.Errors)
+			}
+			if len(timedOut) != 0 {
+				t.Fatalf("cleanup = %v, want none: the replacement is not this invocation's to stop", timedOut)
+			}
+		})
+	}
+}
+
+// TestWaitForStartReadyConfirmsItsOwnRun is the other half of the identity
+// rule: the same row shape, naming the run this invocation started, is ready.
+// A rule that fails every named run would satisfy the test above and break
+// every ordinary wait.
+func TestWaitForStartReadyConfirmsItsOwnRun(t *testing.T) {
+	port, healthz := 19001, "/healthz"
+	path := startManifest(t, "  - name: api\n    cmd: echo api\n")
+	started := "run-r1"
+	startRowsFrom(t, state.Group{Name: "demo", ConfigPath: &path, Services: []state.Service{{
+		Name: "api", Running: true, RunID: &started, PortActual: &port, Health: &healthz,
+		HealthStatus: &state.Health{Status: state.HealthOK},
+	}}})
+
+	chunks := []rpc.GroupsStartChunk{{Service: "api", State: "started", RunID: started, ReadyTimeout: "250ms"}}
+	summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}}
+	if _, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path},
+		nil, chunks, &summary, 0, false); err != nil {
+		t.Fatal(err)
+	}
+	if chunks[0].Ready == nil || !*chunks[0].Ready {
+		t.Fatalf("chunk = %+v, want the run this invocation started reported ready", chunks[0])
+	}
+	if len(summary.Errors) != 0 {
+		t.Fatalf("summary errors = %v, want none", summary.Errors)
+	}
+}
+
+// TestWaitForStartReadyWillNotReadAnUnattributedListener: the row says
+// something is listening and healthy, but it names no run. A listener the row
+// cannot attribute is some process on the service's address, not the run this
+// wait started, so it does not end the wait as ready.
+func TestWaitForStartReadyWillNotReadAnUnattributedListener(t *testing.T) {
+	port, healthz := 19001, "/healthz"
+	path := startManifest(t, "  - name: api\n    cmd: echo api\n")
+	startRowsFrom(t, state.Group{Name: "demo", ConfigPath: &path, Services: []state.Service{{
+		Name: "api", Running: true, PortActual: &port, Health: &healthz,
+		HealthStatus: &state.Health{Status: state.HealthOK},
+	}}})
+
+	chunks := []rpc.GroupsStartChunk{{Service: "api", State: "started", RunID: "run-r1", ReadyTimeout: "100ms"}}
+	summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}}
+	timedOut, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path},
+		nil, chunks, &summary, 0, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if chunks[0].Ready != nil && *chunks[0].Ready {
+		t.Fatal("a listener the row cannot attribute was credited to the run this wait started")
+	}
+	if chunks[0].State != "failed" || chunks[0].Reason != "ready_timeout" {
+		t.Fatalf("chunk = %+v, want the budget to expire unconfirmed", chunks[0])
+	}
+	if len(timedOut) != 1 || timedOut[0].service != "api" || timedOut[0].runID != "run-r1" {
+		t.Fatalf("cleanup = %v, want the run this invocation started", timedOut)
+	}
+}
+
+// TestWaitForStartReadyKeepsLegacyChunksUsable: a daemon that predates run ids
+// names no run to compare against. There is no identity evidence, so the wait
+// keeps reading the service row rather than refusing every old daemon.
+func TestWaitForStartReadyKeepsLegacyChunksUsable(t *testing.T) {
+	port, healthz := 19001, "/healthz"
+	path := startManifest(t, "  - name: api\n    cmd: echo api\n")
+	live := "run-whatever"
+	startRowsFrom(t, state.Group{Name: "demo", ConfigPath: &path, Services: []state.Service{{
+		Name: "api", Running: true, RunID: &live, PortActual: &port, Health: &healthz,
+		HealthStatus: &state.Health{Status: state.HealthOK},
+	}}})
+
+	chunks := []rpc.GroupsStartChunk{{Service: "api", State: "started", ReadyTimeout: "250ms"}}
+	summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}}
+	if _, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path},
+		nil, chunks, &summary, 0, false); err != nil {
+		t.Fatal(err)
+	}
+	if chunks[0].Ready == nil || !*chunks[0].Ready {
+		t.Fatalf("chunk = %+v, want an unnamed run read from the service row", chunks[0])
+	}
+}