test(stack): prove Nacos watch refresh

.github/workflows/ci.yml

@@ -130,11 +130,11 @@ jobs:
       - name: Capture sanitized backend logs
         if: always() && steps.prepare.outcome == 'success'
         shell: bash
-        run: docker compose --project-name "$NEKIRO_E2E_COMPOSE_PROJECT" --file compose.yaml logs --no-color 2>&1 | perl scripts/sanitize-logs.pl
+        run: docker compose --project-name "$NEKIRO_E2E_COMPOSE_PROJECT" --file compose.yaml --profile watch-refresh logs --no-color 2>&1 | perl scripts/sanitize-logs.pl
 
       - name: Tear down backend assembly
         if: always() && steps.prepare.outcome == 'success'
-        run: docker compose --project-name "$NEKIRO_E2E_COMPOSE_PROJECT" --file compose.yaml down --volumes --remove-orphans
+        run: docker compose --project-name "$NEKIRO_E2E_COMPOSE_PROJECT" --file compose.yaml --profile watch-refresh down --volumes --remove-orphans
 
   browser:
     runs-on: ubuntu-latest

.github/workflows/core-integration.yml

@@ -83,11 +83,11 @@ jobs:
       - name: Capture sanitized backend logs
         if: always() && steps.prepare.outcome == 'success'
         shell: bash
-        run: docker compose --project-name "$NEKIRO_E2E_COMPOSE_PROJECT" --file compose.yaml logs --no-color 2>&1 | perl scripts/sanitize-logs.pl
+        run: docker compose --project-name "$NEKIRO_E2E_COMPOSE_PROJECT" --file compose.yaml --profile watch-refresh logs --no-color 2>&1 | perl scripts/sanitize-logs.pl
 
       - name: Tear down backend assembly
         if: always() && steps.prepare.outcome == 'success'
-        run: docker compose --project-name "$NEKIRO_E2E_COMPOSE_PROJECT" --file compose.yaml down --volumes --remove-orphans
+        run: docker compose --project-name "$NEKIRO_E2E_COMPOSE_PROJECT" --file compose.yaml --profile watch-refresh down --volumes --remove-orphans
 
   browser:
     runs-on: ubuntu-latest

components.json

@@ -1,10 +1,10 @@
 {
   "schemaVersion": "1",
-  "contractIdentity": "NeKiro/contracts@6cee25590bb720ed447d70e82204abdad48c5779",
+  "contractIdentity": "NeKiro/contracts@bf705bf0b2cfb412d0c21e46a3b7c0c2bdbe0aa1",
   "components": {
     "core": {
       "repository": "NeKiro-project/NeKiro",
-      "commitSha": "6cee25590bb720ed447d70e82204abdad48c5779"
+      "commitSha": "bf705bf0b2cfb412d0c21e46a3b7c0c2bdbe0aa1"
     },
     "console": {
       "repository": "NeKiro-project/NeKiro-Console",

compose.yaml

@@ -152,6 +152,13 @@ services:
       NEKIRO_ROUTER_NACOS_AUTH_MODE: "none"
       NEKIRO_ROUTER_NACOS_RESPONSE_LIMIT_BYTES: "${NEKIRO_ROUTER_NACOS_RESPONSE_LIMIT_BYTES:?NEKIRO_ROUTER_NACOS_RESPONSE_LIMIT_BYTES must be set and non-empty}"
       NEKIRO_ROUTER_NACOS_REQUEST_TIMEOUT_MS: "${NEKIRO_ROUTER_NACOS_REQUEST_TIMEOUT_MS:?NEKIRO_ROUTER_NACOS_REQUEST_TIMEOUT_MS must be set and non-empty}"
+      NEKIRO_ROUTER_NACOS_OBSERVE_ENABLED: "true"
+      NEKIRO_ROUTER_NACOS_GRPC_TARGET: "nacos:9848"
+      NEKIRO_ROUTER_NACOS_GRPC_CLIENT_IP: "172.28.0.10"
+      NEKIRO_ROUTER_NACOS_GRPC_REQUEST_TIMEOUT_MS: "${NEKIRO_ROUTER_NACOS_GRPC_REQUEST_TIMEOUT_MS:?NEKIRO_ROUTER_NACOS_GRPC_REQUEST_TIMEOUT_MS must be set and non-empty}"
+      NEKIRO_ROUTER_NACOS_PENDING_CHANGES: "${NEKIRO_ROUTER_NACOS_PENDING_CHANGES:?NEKIRO_ROUTER_NACOS_PENDING_CHANGES must be set and non-empty}"
+      NEKIRO_ROUTER_NACOS_MAX_OBSERVATIONS: "${NEKIRO_ROUTER_NACOS_MAX_OBSERVATIONS:?NEKIRO_ROUTER_NACOS_MAX_OBSERVATIONS must be set and non-empty}"
+      NEKIRO_ROUTER_NACOS_GRPC_TRANSPORT_SECURITY: "insecure"
       NEKIRO_ROUTER_CONFIG_CENTER_FILE_ROOT: "/var/lib/nekiro/config"
       NEKIRO_ROUTER_CONFIG_CENTER_MAX_PAYLOAD_BYTES: "${NEKIRO_ROUTER_CONFIG_CENTER_MAX_PAYLOAD_BYTES:?NEKIRO_ROUTER_CONFIG_CENTER_MAX_PAYLOAD_BYTES must be set and non-empty}"
       NEKIRO_ROUTER_INSTANCE_DIRECTORY_KEY: "${NEKIRO_ROUTER_INSTANCE_DIRECTORY_KEY:?NEKIRO_ROUTER_INSTANCE_DIRECTORY_KEY must be set and non-empty}"
@@ -168,8 +175,9 @@ services:
       nacos-bootstrap:
         condition: service_completed_successfully
     networks:
-      - platform-internal
-      - local-access
+      platform-internal:
+        ipv4_address: 172.28.0.10
+      local-access:
     healthcheck:
       test: ["CMD", "wget", "-q", "-O", "/dev/null", "http://127.0.0.1:8081/readyz"]
       interval: 5s
@@ -248,6 +256,48 @@ services:
       retries: 12
       start_period: 5s
 
+  runtime-b-watch-replacement:
+    profiles: ["watch-refresh"]
+    image: "${NEKIRO_RUNTIME_B_IMAGE:?NEKIRO_RUNTIME_B_IMAGE must be set to the prepared immutable image}"
+    environment:
+      RUNTIME_B_LISTEN_ADDR: "0.0.0.0:8092"
+      RUNTIME_B_AGENT_ID: "runtime-b"
+      RUNTIME_B_INSTANCE_ID: "runtime-b-primary"
+      RUNTIME_B_REGISTRATION_MODE: "nacos"
+      RUNTIME_B_NACOS_API_ORIGIN: "http://nacos:8848/nacos"
+      RUNTIME_B_NACOS_NAMESPACE_ID: "nekiro"
+      RUNTIME_B_NACOS_GROUP_NAME: "NEKIRO"
+      RUNTIME_B_NACOS_SERVICE_NAME: "runtime-b"
+      RUNTIME_B_NACOS_CLUSTER_NAME: "DEFAULT"
+      RUNTIME_B_NACOS_ADVERTISED_IP: "172.28.0.13"
+      RUNTIME_B_NACOS_ADVERTISED_PORT: "8092"
+      RUNTIME_B_NACOS_HEARTBEAT_INTERVAL_MS: "2000"
+      RUNTIME_B_NACOS_REQUEST_TIMEOUT_MS: "3000"
+      RUNTIME_B_NACOS_AUTH_MODE: "none"
+      RUNTIME_B_ROUTER_URL: "http://a2a-router:8081"
+      RUNTIME_B_ROUTER_TOKEN: "${RUNTIME_B_ROUTER_TOKEN:?RUNTIME_B_ROUTER_TOKEN must be set and non-empty}"
+      RUNTIME_B_TARGET_AGENT_ID: "runtime-a"
+      RUNTIME_B_TARGET_CAPABILITY: "runtime.echo"
+      RUNTIME_B_RESPONSE_LIMIT_BYTES: "${RUNTIME_B_RESPONSE_LIMIT_BYTES:?RUNTIME_B_RESPONSE_LIMIT_BYTES must be set and non-empty}"
+      RUNTIME_B_EVENT_LIMIT_BYTES: "${RUNTIME_B_EVENT_LIMIT_BYTES:?RUNTIME_B_EVENT_LIMIT_BYTES must be set and non-empty}"
+      NEKIRO_AGENT_CHALLENGE_DIRECTORY: "/tmp/nekiro-challenges"
+      NEKIRO_AGENT_ROUTER_ISSUER: "${NEKIRO_AGENT_ROUTER_ISSUER:?NEKIRO_AGENT_ROUTER_ISSUER must be set and non-empty}"
+      NEKIRO_AGENT_ROUTER_AUDIENCE: "http://runtime-b:8092"
+      NEKIRO_AGENT_ROUTER_KEY_ID: "${NEKIRO_AGENT_ROUTER_KEY_ID:?NEKIRO_AGENT_ROUTER_KEY_ID must be set and non-empty}"
+      NEKIRO_AGENT_ROUTER_PUBLIC_KEY_BASE64URL: "${NEKIRO_AGENT_ROUTER_PUBLIC_KEY_BASE64URL:?NEKIRO_AGENT_ROUTER_PUBLIC_KEY_BASE64URL must be set and non-empty}"
+    depends_on:
+      a2a-router:
+        condition: service_healthy
+    networks:
+      platform-internal:
+        ipv4_address: 172.28.0.13
+    healthcheck:
+      test: ["CMD", "wget", "-q", "-O", "/dev/null", "http://127.0.0.1:8092/readyz"]
+      interval: 5s
+      timeout: 5s
+      retries: 12
+      start_period: 5s
+
   runtime-a:
     image: "${NEKIRO_RUNTIME_A_IMAGE:?NEKIRO_RUNTIME_A_IMAGE must be set to the prepared immutable image}"
     environment:

scripts/write-ci-env.sh

@@ -53,6 +53,9 @@ NEKIRO_ROUTER_INSTANCE_DIRECTORY_KEY=router/instance-directory
 NEKIRO_ROUTER_INSTANCE_PORT_NAME=a2a
 NEKIRO_ROUTER_NACOS_RESPONSE_LIMIT_BYTES=1048576
 NEKIRO_ROUTER_NACOS_REQUEST_TIMEOUT_MS=3000
+NEKIRO_ROUTER_NACOS_GRPC_REQUEST_TIMEOUT_MS=3000
+NEKIRO_ROUTER_NACOS_PENDING_CHANGES=64
+NEKIRO_ROUTER_NACOS_MAX_OBSERVATIONS=1024
 EOF
   printf 'NEKIRO_ROUTER_CONFIG_CENTER_ROOT=%s\n' "$config_root" >>"$output"
   printf 'NEKIRO_E2E_COMPOSE_FILE=%s/compose.yaml\n' "$stack_root" >>"$output"

tests/backend/invoke_record_test.go

@@ -118,6 +118,7 @@ func TestInvokeToRecordAcceptance(t *testing.T) {
 		"policy-content-secret", "protocol-content-secret", "agent-content-secret",
 		"route-content-secret", "timeout-content-secret", "cancel-content-secret",
 		"interrupted-content-secret", "dependency-content-secret", "dependency-raw-secret",
+		"snapshot-refresh-value",
 	}, env.credentialForbidden...)
 	client := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, Timeout: 45 * time.Second}
 	if result := doRequest(t, client, env.controlPlane+"/readyz", http.MethodGet, "", "", nil); result.status != http.StatusNoContent {
@@ -181,6 +182,15 @@ func TestInvokeToRecordAcceptance(t *testing.T) {
 	}
 	assertRecord(t, client, env, stream[0].InvocationID, acceptanceWorkspace, "runtime-b", "succeeded", "")
 
+	replaceRuntimeBNacosInstance(t, client, env)
+	runtimeBReplacementResult := invokeJSON(t, client, env, "runtime-b", "runtime.echo", map[string]any{"fixture": "success", "value": "snapshot-refresh-value"})
+	if runtimeBReplacementResult.result.Status != "succeeded" ||
+		!bytes.Contains(runtimeBReplacementResult.result.Result, []byte("snapshot-refresh-value")) ||
+		!bytes.Contains(runtimeBReplacementResult.result.Result, []byte(`"instanceId":"runtime-b-primary"`)) {
+		t.Fatalf("Runtime B replacement JSON result=%s", runtimeBReplacementResult.result.Result)
+	}
+	assertRecord(t, client, env, runtimeBReplacementResult.result.InvocationID, acceptanceWorkspace, "runtime-b", "succeeded", "")
+
 	nested := invokeJSON(t, client, env, "runtime-a", "runtime.cross", map[string]any{"fixture": "success", "value": "nested-value"})
 	if nested.result.Status != "succeeded" || !bytes.Contains(nested.result.Result, []byte(`"runtime-a"`)) || !bytes.Contains(nested.result.Result, []byte(`"childInvocationId"`)) {
 		t.Fatalf("nested result=%s", nested.result.Result)
@@ -298,6 +308,61 @@ func assertNacosRegistrations(t *testing.T, client *http.Client, env acceptanceE
 	}
 }
 
+func replaceRuntimeBNacosInstance(t *testing.T, client *http.Client, env acceptanceEnv) {
+	t.Helper()
+	stop := composeCommand(t.Context(), env, "stop", "runtime-b-directory")
+	if output, err := stop.CombinedOutput(); err != nil {
+		t.Fatalf("stop original Runtime B directory instance: %v output=%s", err, output)
+	}
+
+	start := composeCommand(
+		t.Context(), env,
+		"--profile", "watch-refresh", "up", "--detach", "--no-deps", "--force-recreate",
+		"--wait", "--wait-timeout", "60", "runtime-b-watch-replacement",
+	)
+	if output, err := start.CombinedOutput(); err != nil {
+		t.Fatalf("start replacement Runtime B directory instance: %v output=%s", err, output)
+	}
+	waitForNacosInstance(t, client, env, "runtime-b", "runtime-b-primary")
+}
+
+func waitForNacosInstance(t *testing.T, client *http.Client, env acceptanceEnv, serviceName, instanceID string) {
+	t.Helper()
+	endpoint, err := url.Parse(env.nacosURL + "/v1/ns/instance/list")
+	if err != nil {
+		t.Fatal(err)
+	}
+	query := endpoint.Query()
+	query.Set("serviceName", serviceName)
+	query.Set("groupName", "NEKIRO")
+	query.Set("clusters", "DEFAULT")
+	query.Set("namespaceId", "nekiro")
+	query.Set("healthyOnly", "false")
+	endpoint.RawQuery = query.Encode()
+
+	deadline := time.Now().Add(30 * time.Second)
+	var last httpResult
+	for time.Now().Before(deadline) {
+		last = doRequest(t, client, endpoint.String(), http.MethodGet, "", "", nil)
+		var response struct {
+			Hosts []struct {
+				Healthy   bool              `json:"healthy"`
+				Enabled   bool              `json:"enabled"`
+				Ephemeral bool              `json:"ephemeral"`
+				Metadata  map[string]string `json:"metadata"`
+			} `json:"hosts"`
+		}
+		if last.status == http.StatusOK && json.Unmarshal(last.body, &response) == nil && len(response.Hosts) == 1 {
+			host := response.Hosts[0]
+			if host.Healthy && host.Enabled && host.Ephemeral && host.Metadata["nekiro.instanceId"] == instanceID {
+				return
+			}
+		}
+		time.Sleep(250 * time.Millisecond)
+	}
+	t.Fatalf("Nacos service %s did not converge to instance %s: status=%d body=%s", serviceName, instanceID, last.status, last.body)
+}
+
 func publishRouterNacosBindings(t *testing.T, client *http.Client, env acceptanceEnv) {
 	t.Helper()
 	agentIDs := make([]string, 0, len(env.releases))
@@ -1846,7 +1911,7 @@ func assertStorageAndLogsAreMetadataOnly(t *testing.T, env acceptanceEnv) {
 		t.Fatal(err)
 	}
 	installationRows.Close()
-	logs := composeCommand(ctx, env, "logs", "--no-color")
+	logs := composeCommand(ctx, env, "--profile", "watch-refresh", "logs", "--no-color")
 	output, err := logs.Output()
 	if err != nil {
 		t.Fatal(err)