fix(engine): up --wait 超时清理按运行身份停止

docs/agent/architecture.md

@@ -120,6 +120,13 @@ A start accepts only readiness evidence for the current service. The no-port run
 the stop receipt are fixed before the first signal and are not re-derived from the same PID in a later
 stage. Operations are refused when reliable control identity is missing.
 
+A readiness timeout's cleanup is one of those operations: it names the run this invocation started for
+each service (`expect_runs` on `groups.kill`), and the daemon selects only listeners and no-port runs
+whose run id matches. A service another controller restarted during the wait is a different run and is
+left alone; a target whose identity cannot be verified is not selected either. The CLI refuses to send
+a name-only cleanup to a daemon that does not announce the capability, because that daemon would stop
+whatever answers to the name.
+
 `down` captures reservations before signalling, reuses a fresh scan after the actual stop, checks the
 original registry (including compatibility stopping records), and deletes only after comparing the full
 reservation set inside a short transaction. Failure, cancellation, partial stops, remaining runs or

docs/cli.md

@@ -477,8 +477,12 @@ it exits 2 without running the start; a service that times out has state `failed
 `ready_timeout`, the message says whether the timeout came from that service's readiness budget or
 from the caller's `--wait-timeout` ceiling, and the command exits with code 1. Only runs newly started
 this time that time out enter stop cleanup; existing runs are not stopped or restarted because this
-wait failed. 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
+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` /
+`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
 `status`, the desktop service row, attention and the event stream do not show it as an ordinary manual
 stop. Health facts carry `status`, `code`, `reason` and `observed_at`; when a dependency declared

docs/cli.zh.md

@@ -397,7 +397,10 @@ cwd 中执行，按 `cmd:` 的规则拆成 argv（不经过 shell），支持 `$
 启用 `--wait` 或 `--ready`，否则退出 2，不执行启动；
 超时的服务状态为 `failed`、原因是 `ready_timeout`，消息说明超时来自该服务的就绪预算，还是调用方给的
 `--wait-timeout` 上限，命令退出码为 1。只有本次新启动的超时运行才进入停止清理；
-已经存在的运行不会因这次等待失败而被停止或重启。`started` / `skipped` 汇总描述启动动作，
+已经存在的运行不会因这次等待失败而被停止或重启。清理按 `run_id` 只停止本次启动的那个运行：
+等待期间另一个控制者为同一服务启动的新运行是另一个运行，不会被停掉。daemon 若未声明
+`groups.kill.expect_runs`，就无法把停止限定到指定运行，因此不会收到只带服务名的清理：
+命令失败，并说明超时的运行仍在运行。`started` / `skipped` 汇总描述启动动作，
 就绪判定失败另外出现在 `errors` 与逐服务 `state` 中。对本次新启动且被清理的运行，
 引擎把同一个 `ready_timeout` 写入服务的最后退出事实，因此 `status`、桌面服务行、attention 和事件流
 不会把它显示成普通的手动停止。健康事实带有 `status`、`code`、`reason` 与 `observed_at`，

engine/docs/schema/protocol.schema.json

@@ -842,6 +842,12 @@
           },
           "type": "array"
         },
+        "expect_runs": {
+          "items": {
+            "$ref": "#/definitions/RunExpectation"
+          },
+          "type": "array"
+        },
         "release": {
           "type": "boolean"
         },
@@ -1683,6 +1689,21 @@
         "root_pid"
       ]
     },
+    "RunExpectation": {
+      "properties": {
+        "service": {
+          "type": "string"
+        },
+        "run_id": {
+          "type": "string"
+        }
+      },
+      "type": "object",
+      "required": [
+        "service",
+        "run_id"
+      ]
+    },
     "RunRecord": {
       "properties": {
         "id": {

engine/internal/cmd/ready_timeout_test.go

@@ -94,18 +94,22 @@ func TestWaitForStartReadyUsesEachServiceBudget(t *testing.T) {
 	chdir(t, dir)
 
 	chunks := []rpc.GroupsStartChunk{
-		{Service: "api", State: "started", ReadyTimeout: "15ms"},
-		{Service: "worker", State: "started", ReadyTimeout: "35ms"},
+		{Service: "api", State: "started", ReadyTimeout: "15ms", RunID: "run-api"},
+		{Service: "worker", State: "started", ReadyTimeout: "35ms", RunID: "run-worker"},
 	}
 	summary := rpc.GroupsStartEnd{Started: []string{"api", "worker"}}
 	timedOut, err := waitForStartReady(context.Background(),
 		rpc.GroupsStartParams{ConfigPath: &path}, nil, chunks, &summary, 0, false)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(timedOut) != 2 || timedOut[0] != "api" || timedOut[1] != "worker" {
+	if len(timedOut) != 2 || timedOut[0].service != "api" || timedOut[1].service != "worker" {
 		t.Fatalf("timed out = %v, want both owned services", timedOut)
 	}
+	// The cleanup is authorized by run identity, not by the service name.
+	if timedOut[0].runID != "run-api" || timedOut[1].runID != "run-worker" {
+		t.Fatalf("timed out = %v, want each service's own run", timedOut)
+	}
 	byName := map[string]string{}
 	for _, chunk := range chunks {
 		byName[chunk.Service] = chunk.Error

engine/internal/cmd/timeout_cleanup_test.go

@@ -0,0 +1,100 @@
+package cmd
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/sheathedsharp/option-berth/engine/internal/daemon/rpc"
+)
+
+// fakeKillCaller records the one call a timeout cleanup is allowed to make.
+type fakeKillCaller struct {
+	hello rpc.DaemonHelloResult
+	calls []struct {
+		method string
+		params any
+	}
+}
+
+func (f *fakeKillCaller) Call(_ context.Context, method string, params, result any) error {
+	f.calls = append(f.calls, struct {
+		method string
+		params any
+	}{method: method, params: params})
+	if env, ok := result.(*rpc.KillEnvelope); ok {
+		env.MutationResult.OK = true
+	}
+	return nil
+}
+
+func (f *fakeKillCaller) Hello() rpc.DaemonHelloResult { return f.hello }
+
+// The cleanup stops the runs this invocation started, not whatever now answers
+// to the service's name; that is the only thing that makes another controller's
+// replacement run safe from this timeout.
+func TestStopTimedOutServicesNamesTheRunsItStarted(t *testing.T) {
+	caller := &fakeKillCaller{hello: rpc.DaemonHelloResult{
+		Capabilities: []string{rpc.CapabilityKillExpectRuns},
+	}}
+	timedOut := []startTimeout{{service: "api", runID: "run-api"}, {service: "worker", runID: "run-worker"}}
+
+	if err := stopTimedOutServices(context.Background(), caller, rpc.GroupsStartParams{}, timedOut); err != nil {
+		t.Fatalf("stopTimedOutServices: %v", err)
+	}
+	if len(caller.calls) != 1 || caller.calls[0].method != "groups.kill" {
+		t.Fatalf("calls = %+v, want one groups.kill", caller.calls)
+	}
+	params, ok := caller.calls[0].params.(rpc.GroupsKillParams)
+	if !ok {
+		t.Fatalf("params = %T, want rpc.GroupsKillParams", caller.calls[0].params)
+	}
+	if len(params.Only) != 0 {
+		t.Fatalf("only = %v, must be empty: a name-scoped stop can reach a replacement run", params.Only)
+	}
+	if len(params.ExpectRuns) != 2 ||
+		params.ExpectRuns[0] != (rpc.RunExpectation{Service: "api", RunID: "run-api"}) ||
+		params.ExpectRuns[1] != (rpc.RunExpectation{Service: "worker", RunID: "run-worker"}) {
+		t.Fatalf("expect_runs = %+v, want both started runs", params.ExpectRuns)
+	}
+	if !params.Release || params.Reason != "ready_timeout" {
+		t.Fatalf("params = %+v, want the readiness-timeout stop", params)
+	}
+}
+
+// A daemon that predates expect_runs ignores it and stops by name. Sending the
+// cleanup anyway is how it reaches the run somebody else started, so the stop
+// is refused instead — and the error says the timed-out runs are still up.
+func TestStopTimedOutServicesRefusesADaemonThatStopsByName(t *testing.T) {
+	caller := &fakeKillCaller{hello: rpc.DaemonHelloResult{DaemonVersion: "0.9.0"}}
+
+	err := stopTimedOutServices(context.Background(), caller,
+		rpc.GroupsStartParams{}, []startTimeout{{service: "api", runID: "run-api"}})
+	if err == nil {
+		t.Fatal("a daemon without expect_runs must not receive a name-only cleanup")
+	}
+	if !strings.Contains(err.Error(), "api") || !strings.Contains(err.Error(), "not stopped") ||
+		!strings.Contains(err.Error(), "daemon restart") {
+		t.Fatalf("error = %v, want the services, what was not done, and the way out", err)
+	}
+	if len(caller.calls) != 0 {
+		t.Fatalf("calls = %+v, want no stop at all", caller.calls)
+	}
+}
+
+// Same rule for a run this invocation cannot name: without the id there is no
+// way to tell the timed-out run from its replacement, so nothing is signalled.
+func TestStopTimedOutServicesRefusesAnUnnamedRun(t *testing.T) {
+	caller := &fakeKillCaller{hello: rpc.DaemonHelloResult{
+		Capabilities: []string{rpc.CapabilityKillExpectRuns},
+	}}
+
+	err := stopTimedOutServices(context.Background(), caller,
+		rpc.GroupsStartParams{}, []startTimeout{{service: "api"}})
+	if err == nil || !strings.Contains(err.Error(), "api") {
+		t.Fatalf("error = %v, want a refusal naming the service", err)
+	}
+	if len(caller.calls) != 0 {
+		t.Fatalf("calls = %+v, want no stop at all", caller.calls)
+	}
+}

engine/internal/cmd/up.go

@@ -263,15 +263,46 @@ func consumeStartWithWait(ctx context.Context, c *client.Client, stream *client.
 	return nil
 }
 
+// killCaller is the part of the daemon client a timeout cleanup needs: one
+// call and the capability list. It is an interface so the stop request can be
+// tested without a live daemon.
+type killCaller interface {
+	Call(ctx context.Context, method string, params, result any) error
+	Hello() rpc.DaemonHelloResult
+}
+
 // stopTimedOutServices makes a readiness timeout terminal. The same group
 // mutation path used by `oberth down` stops portless workers as well as
-// listeners, while Only keeps already-running or skipped services untouched.
-func stopTimedOutServices(ctx context.Context, c *client.Client, params rpc.GroupsStartParams, services []string) error {
+// listeners, but the stop names the run this invocation started for each
+// service. A service name alone would also match a run another controller
+// started while this wait was running — that run is not ours to stop and must
+// survive the timeout. `expect_runs` also keeps the stop scoped, so
+// already-running or skipped services and their `port: auto` claims stay.
+//
+// A daemon that does not know `expect_runs` would silently drop the field and
+// stop by name, which is exactly the wrong run; the cleanup is refused then,
+// and an invocation that cannot name the run it started refuses too.
+func stopTimedOutServices(ctx context.Context, c killCaller, params rpc.GroupsStartParams, timedOut []startTimeout) error {
+	if !hasCapability(c.Hello(), rpc.CapabilityKillExpectRuns) {
+		return timeoutCleanupRefused(c.Hello(), timedOut,
+			"does not announce "+rpc.CapabilityKillExpectRuns+", so it would stop whatever answers to the service name")
+	}
+	expect := make([]rpc.RunExpectation, 0, len(timedOut))
+	for _, timedOutService := range timedOut {
+		if timedOutService.runID == "" {
+			return timeoutCleanupRefused(c.Hello(), timedOut,
+				"did not name the run it started for "+timedOutService.service)
+		}
+		expect = append(expect, rpc.RunExpectation{
+			Service: timedOutService.service,
+			RunID:   timedOutService.runID,
+		})
+	}
 	var env rpc.KillEnvelope
 	err := c.Call(ctx, "groups.kill", rpc.GroupsKillParams{
 		Name:       valueOrEmpty(params.Name),
 		ConfigPath: params.ConfigPath,
-		Only:       services,
+		ExpectRuns: expect,
 		Release:    true,
 		Reason:     "ready_timeout",
 	}, &env)
@@ -281,6 +312,22 @@ func stopTimedOutServices(ctx context.Context, c *client.Client, params rpc.Grou
 	return nil
 }
 
+// timeoutCleanupRefused says plainly that the timed-out runs were left running
+// and why: a cleanup that stops the wrong run must not read as success, and an
+// agent needs the state it is actually in.
+func timeoutCleanupRefused(hello rpc.DaemonHelloResult, timedOut []startTimeout, why string) error {
+	version := hello.DaemonVersion
+	if version == "" {
+		version = "an older version"
+	}
+	services := make([]string, 0, len(timedOut))
+	for _, service := range timedOut {
+		services = append(services, service.service)
+	}
+	return fmt.Errorf("the timed-out runs for %s were not stopped: the running daemon (%s) %s\nhint: they may still be running; stop them with `oberth down`, or restart the daemon with `oberth daemon restart`",
+		strings.Join(services, ", "), version, why)
+}
+
 func valueOrEmpty(value *string) string {
 	if value == nil {
 		return ""

engine/internal/cmd/up_ready.go

@@ -14,14 +14,23 @@ import (
 	"github.com/sheathedsharp/option-berth/engine/internal/state"
 )
 
+// 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
+// started while this wait was still running.
+type startTimeout struct {
+	service string
+	runID   string
+}
+
 // waitForStartReady waits on the same state that status exposes. A service
 // with a configured health path is ready only after its listener reports ok;
 // every other service is ready once its run is live (or its listener is up).
 // This keeps `up --wait` useful to both humans and agents without adding a
 // second readiness protocol to the daemon.
 func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *groups.Config,
 	chunks []rpc.GroupsStartChunk, summary *rpc.GroupsStartEnd, timeout time.Duration,
-	timeoutSet bool) ([]string, error) {
+	timeoutSet bool) ([]startTimeout, error) {
 	waitStart := time.Now()
 	// An explicit --wait-timeout is the ceiling for the whole invocation. It
 	// does not replace each service's own readiness budget; it only caps the
@@ -51,7 +60,7 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g
 	// start would read an exit from this same second as older than the wait.
 	// Align the timestamp fallback to the second; the run id stays primary.
 	fallbackSince := waitStart.UTC().Truncate(time.Second)
-	var timedOut []string
+	var timedOut []startTimeout
 	for {
 		pp, gg, obs, err := groupRowsObserved(ctx)
 		if obs != nil && obs.State != string(ports.ObservationObservable) {
@@ -94,8 +103,12 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g
 				}
 				// An older daemon may omit run_id on a skipped chunk. Once
 				// the service is live, the group row names its run, and
-				// that run is what this wait is about.
-				if selection.runID == "" && svc.Running && svc.RunID != nil {
+				// that run is what this wait is about. Only a run this
+				// invocation did not start is read from the live row: for
+				// its own start the chunk is the authority, and a run id
+				// copied from whatever is live now could be a replacement
+				// that the timeout cleanup must not touch.
+				if selection.runID == "" && !selection.owned && svc.Running && svc.RunID != nil {
 					selection.runID = *svc.RunID
 					selected[name] = selection
 				}
@@ -117,7 +130,7 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g
 			}
 		}
 		if len(selected) == 0 {
-			sort.Strings(timedOut)
+			sortStartTimeouts(timedOut)
 			return timedOut, nil
 		}
 		now := time.Now()
@@ -129,12 +142,12 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g
 				startTimeoutMessage(selection, name, timeout), nil)
 			// Waiting on an existing run does not authorize stopping it.
 			if selection.owned {
-				timedOut = append(timedOut, name)
+				timedOut = append(timedOut, startTimeout{service: name, runID: selection.runID})
 			}
 			delete(selected, name)
 		}
 		if len(selected) == 0 {
-			sort.Strings(timedOut)
+			sortStartTimeouts(timedOut)
 			return timedOut, nil
 		}
 		select {
@@ -195,6 +208,12 @@ func startChunkReadyTimeout(c rpc.GroupsStartChunk) time.Duration {
 	return groups.DefaultReadyTimeout
 }
 
+// sortStartTimeouts orders the cleanup list by service, so the stop request
+// and its failure message read the same way on every run.
+func sortStartTimeouts(timeouts []startTimeout) {
+	sort.Slice(timeouts, func(i, j int) bool { return timeouts[i].service < timeouts[j].service })
+}
+
 // startSelection is what this invocation selected for one service: the run it
 // is about, whether this invocation started that run (and may therefore stop
 // it on timeout), the readiness budget its manifest gives it, and the deadline

engine/internal/cmd/up_test.go

@@ -166,7 +166,7 @@ func TestWaitForStartReadyUsesTimeoutAsStructuredFailure(t *testing.T) {
 	summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}}
 	if timedOut, err := waitForStartReady(context.Background(), params, nil, chunks, &summary, time.Millisecond, true); err != nil {
 		t.Fatal(err)
-	} else if len(timedOut) != 1 || timedOut[0] != "api" {
+	} else if len(timedOut) != 1 || timedOut[0].service != "api" {
 		t.Fatalf("timed out = %v, want api", timedOut)
 	}
 	if chunks[0].State != "failed" || chunks[0].Reason != "ready_timeout" ||
@@ -415,14 +415,14 @@ func TestWaitForStartReadyMixedTimeoutCleansOnlyNewStarts(t *testing.T) {
 		t.Fatal(err)
 	}
 	chdir(t, dir)
-	chunks := []rpc.GroupsStartChunk{{Service: "old", State: "skipped", Skipped: true}, {Service: "new", State: "started"}}
+	chunks := []rpc.GroupsStartChunk{{Service: "old", State: "skipped", Skipped: true}, {Service: "new", State: "started", RunID: "run-new"}}
 	summary := rpc.GroupsStartEnd{Skipped: []string{"old"}, Started: []string{"new"}}
 	cleanup, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path}, nil, chunks, &summary, time.Millisecond, true)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(cleanup) != 1 || cleanup[0] != "new" {
-		t.Fatalf("cleanup=%v, must exclude existing service", cleanup)
+	if len(cleanup) != 1 || cleanup[0].service != "new" || cleanup[0].runID != "run-new" {
+		t.Fatalf("cleanup=%v, must be the run this invocation started and only it", cleanup)
 	}
 	if len(summary.Errors) != 2 {
 		t.Fatalf("both services must fail readiness: %+v", summary)