Remove Continue-As-New and InternalHandlerTimeout; rename FlowExecutionID to FlowRunID (#1541)

Author: longquanzheng

.cursor/rules/project-core.mdc

@@ -173,9 +173,7 @@ still not describe downstream products.
 ## User-Facing Concept Boundary
 
 User-facing docs, examples and skills describe Dex concepts and guarantees only.
-Do not mention Continue-As-New except in the Flow options reference for the
-Continue-As-New threshold and the TriggerContinueAsNew Client API reference. Do
-not describe server-zd engine internals (apply, commands, partitions, history
+Do not describe server-zd engine internals (apply, commands, partitions, history
 mechanics) as part of the application model; state the Dex behavior instead.
 Deployment and operations guidance may name server-zd and its configuration.
 

AGENTS.md

@@ -322,9 +322,7 @@ still not describe downstream products.
 ### User-Facing Concept Boundary
 
 User-facing docs, examples and skills describe Dex concepts and guarantees only.
-Do not mention Continue-As-New except in the Flow options reference for the
-Continue-As-New threshold and the TriggerContinueAsNew Client API reference. Do
-not describe server-zd engine internals (apply, commands, partitions, history
+Do not describe server-zd engine internals (apply, commands, partitions, history
 mechanics) as part of the application model; state the Dex behavior instead.
 Deployment and operations guidance may name server-zd and its configuration.
 

CLAUDE.md

@@ -245,9 +245,7 @@ still not describe downstream products.
 ## User-Facing Concept Boundary
 
 User-facing docs, examples and skills describe Dex concepts and guarantees only.
-Do not mention Continue-As-New except in the Flow options reference for the
-Continue-As-New threshold and the TriggerContinueAsNew Client API reference. Do
-not describe server-zd engine internals (apply, commands, partitions, history
+Do not describe server-zd engine internals (apply, commands, partitions, history
 mechanics) as part of the application model; state the Dex behavior instead.
 Deployment and operations guidance may name server-zd and its configuration.
 

cli/README.md

@@ -213,7 +213,7 @@ dexcli health
 dexcli flow start order-123 --flow-type OrderFlow --start-step-type StartOrder --input '{"order":123}' --yes
 dexcli flow search --query 'ExecutionStatus = "Running"'
 dexcli flow inspect order-123 --all-history
-dexcli flow watch order-123 --follow-runs
+dexcli flow watch order-123
 ```
 
 JSON is the default and stable machine-readable output. Use `--output table`
@@ -643,7 +643,7 @@ dexcli flow state FLOW_ID [--run-id RUN_ID]
 dexcli flow history FLOW_ID [--run-id RUN_ID] [--start-event-id N]
                     [--page-size N] [--page-token BASE64] [--all]
 dexcli flow inspect FLOW_ID [--run-id RUN_ID] [--all-history]
-dexcli flow watch FLOW_ID [--run-id RUN_ID] [--from-event-id N] [--follow-runs]
+dexcli flow watch FLOW_ID [--run-id RUN_ID] [--from-event-id N]
 dexcli flow stop FLOW_ID [--run-id RUN_ID]
                  [--type cancel|terminate|fail] [--reason TEXT] --yes
 dexcli flow skip-timer FLOW_ID --step-type TYPE [--execution N]
@@ -657,10 +657,8 @@ dexcli flow time-travel FLOW_ID [--run-id RUN_ID]
 ```
 
 With the default JSON output, `watch` writes one object per line and exits when
-the run becomes terminal.
-`--follow-runs` keeps watching the Flow's current run when it changes. Stop and
-time travel operate on the current run by default; pass `--run-id` to target an
-exact run. All mutating Flow commands require `--yes`, including in
+the run becomes terminal. Stop and time travel operate on the current run by
+default; pass `--run-id` to target an exact run. All mutating Flow commands require `--yes`, including in
 non-interactive use.
 `--step-method` is required with `--type step-execution-id` and invalid for other types.
 

cli/internal/command/api.go

@@ -240,7 +240,7 @@ func fullMethodName(service protoreflect.ServiceDescriptor, method protoreflect.
 func isMutatingMethod(name protoreflect.Name) bool {
 	switch name {
 	case "StartFlow", "PublishToChannel", "WriteStream", "StopFlow", "SetAttributes", "SyncAttributeIndexes",
-		"ResetFlow", "InvokeRPC", "SkipTimer", "UpdateFlowConfig", "TriggerContinueAsNew", "DeleteChannelMessage":
+		"ResetFlow", "InvokeRPC", "SkipTimer", "UpdateFlowConfig", "DeleteChannelMessage":
 		return true
 	default:
 		return false

cli/internal/command/command_integ_test.go

@@ -30,30 +30,28 @@ import (
 type testFlowService struct {
 	dexpb.UnimplementedFlowServiceServer
 
-	mu                   sync.Mutex
-	loadCalls            int
-	loadRequest          *dexpb.LoadBlobsRequest
-	startCalls           int
-	stopCalls            int
-	streamWriteCalls     int
-	skipTimerCalls       int
-	waitStepCalls        int
-	updateConfigCalls    int
-	continueAsNewCalls   int
-	startRequest         *dexpb.StartFlowRequest
-	stopRequest          *dexpb.StopFlowRequest
-	streamWriteRequest   *dexpb.WriteStreamRequest
-	waitRequest          *dexpb.WaitForFlowRequest
-	skipTimerRequest     *dexpb.SkipTimerRequest
-	waitStepRequest      *dexpb.WaitForStepCompletionRequest
-	updateConfigRequest  *dexpb.UpdateFlowConfigRequest
-	continueAsNewRequest *dexpb.TriggerContinueAsNewRequest
-	timeTravelCalls      int
-	timeTravelRequest    *dexpb.ResetFlowRequest
-	waitStarted          chan struct{}
-	waitOnce             sync.Once
-	serverInfo           *dexpb.ServerInfo
-	serverInfoFailure    error
+	mu                  sync.Mutex
+	loadCalls           int
+	loadRequest         *dexpb.LoadBlobsRequest
+	startCalls          int
+	stopCalls           int
+	streamWriteCalls    int
+	skipTimerCalls      int
+	waitStepCalls       int
+	updateConfigCalls   int
+	startRequest        *dexpb.StartFlowRequest
+	stopRequest         *dexpb.StopFlowRequest
+	streamWriteRequest  *dexpb.WriteStreamRequest
+	waitRequest         *dexpb.WaitForFlowRequest
+	skipTimerRequest    *dexpb.SkipTimerRequest
+	waitStepRequest     *dexpb.WaitForStepCompletionRequest
+	updateConfigRequest *dexpb.UpdateFlowConfigRequest
+	timeTravelCalls     int
+	timeTravelRequest   *dexpb.ResetFlowRequest
+	waitStarted         chan struct{}
+	waitOnce            sync.Once
+	serverInfo          *dexpb.ServerInfo
+	serverInfoFailure   error
 }
 
 func (s *testFlowService) GetServerInfo(
@@ -188,17 +186,6 @@ func (s *testFlowService) UpdateFlowConfig(
 	return &emptypb.Empty{}, nil
 }
 
-func (s *testFlowService) TriggerContinueAsNew(
-	_ context.Context,
-	request *dexpb.TriggerContinueAsNewRequest,
-) (*emptypb.Empty, error) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.continueAsNewCalls++
-	s.continueAsNewRequest = request
-	return &emptypb.Empty{}, nil
-}
-
 func (s *testFlowService) ResetFlow(
 	_ context.Context,
 	request *dexpb.ResetFlowRequest,
@@ -219,8 +206,8 @@ func (s *testFlowService) GetFlowSummary(
 		flowStatus = dexpb.FlowStatus_FLOW_STATUS_COMPLETED
 	}
 	return &dexpb.GetFlowSummaryResponse{
-		FlowExecutionId: &dexpb.FlowExecutionID{FlowId: request.GetFlowId(), RunId: "run-1"},
-		FirstRunId:      "run-1", FlowType: "OrderFlow", FlowStatus: flowStatus,
+		FlowRunId:  &dexpb.FlowRunID{FlowId: request.GetFlowId(), RunId: "run-1"},
+		FirstRunId: "run-1", FlowType: "OrderFlow", FlowStatus: flowStatus,
 	}, nil
 }
 
@@ -403,7 +390,7 @@ func TestFlowClientOperationsMapSDKEquivalentRequests(t *testing.T) {
 	startResult := executeTestCommand(t, nil,
 		"flow", "start", "flow-1", "--flow-type", "OrderFlow", "--start-step-type", "StartOrder",
 		"--input", `{"order":42}`, "--attributes", `[{"key":"status","value":"new","index":{"type":"keyword"},"sync":true}]`,
-		"--config", `{"activeStepSearchMode":"all","continueAsNewThreshold":12,"stepDurability":"sync","offloadHistory":false}`,
+		"--config", `{"activeStepSearchMode":"all","stepDurability":"sync","offloadHistory":false}`,
 		"--retry-policy", `{"initialInterval":"1500ms","backoffCoefficient":2,"maximumInterval":"5s","maximumAttempts":3}`,
 		"--step-options", `{"skipWaitFor":true}`, "--timeout-handler-options", `{"methodTimeoutSeconds":20,"loadChannelNames":["commands"]}`,
 		"--flow-timeout", "90s", "--flow-timeout-policy", "handler",
@@ -424,8 +411,7 @@ func TestFlowClientOperationsMapSDKEquivalentRequests(t *testing.T) {
 
 	executeTestCommand(t, nil, "flow", "skip-timer", "flow-1", "--step-type", "WaitForPayment", "--execution", "2", "--condition-index", "3", "--yes", "--server", address)
 	executeTestCommand(t, nil, "flow", "wait-step", "flow-1", "--step-type", "ShipOrder", "--execution", "2", "--request-timeout", "1500ms", "--server", address)
-	executeTestCommand(t, nil, "flow", "update-config", "flow-1", "--config", `{"continueAsNewPageSizeInBytes":1024,"attributeStoreNames":[],"historyWindowEvents":5000,"deleteExpiredHistoryObjects":false}`, "--yes", "--server", address)
-	executeTestCommand(t, nil, "flow", "trigger-continue-as-new", "flow-1", "--yes", "--server", address)
+	executeTestCommand(t, nil, "flow", "update-config", "flow-1", "--config", `{"attributeStoreNames":[],"historyWindowEvents":5000,"deleteExpiredHistoryObjects":false}`, "--yes", "--server", address)
 
 	service.mu.Lock()
 	defer service.mu.Unlock()
@@ -459,12 +445,10 @@ func TestFlowClientOperationsMapSDKEquivalentRequests(t *testing.T) {
 	if service.waitStepRequest.GetStepExecutionNumber() != "2" || service.waitStepRequest.GetRequestTimeoutSeconds() != 2 || service.waitStepRequest.GetRequestId() == "" {
 		t.Fatalf("unexpected wait-step request: %#v", service.waitStepRequest)
 	}
-	if service.updateConfigRequest.GetFlowConfig().GetContinueAsNewPageSizeInBytes() != 1024 || service.updateConfigRequest.GetFlowConfig().AttributeStoreNames == nil || len(service.updateConfigRequest.GetFlowConfig().GetAttributeStoreNames().GetNames()) != 0 || service.updateConfigRequest.GetFlowConfig().GetHistoryWindowEvents() != 5000 || service.updateConfigRequest.GetFlowConfig().DeleteExpiredHistoryObjects != nil || service.updateConfigRequest.GetFlowConfig().OffloadHistory != nil {
+	if service.updateConfigRequest.GetFlowConfig().AttributeStoreNames == nil || len(service.updateConfigRequest.GetFlowConfig().GetAttributeStoreNames().GetNames()) != 0 || service.updateConfigRequest.GetFlowConfig().GetHistoryWindowEvents() != 5000 || service.updateConfigRequest.GetFlowConfig().DeleteExpiredHistoryObjects != nil || service.updateConfigRequest.GetFlowConfig().OffloadHistory != nil {
 		t.Fatalf("unexpected update-config request: %#v", service.updateConfigRequest)
 	}
-	if service.continueAsNewRequest.GetFlowId() != "flow-1" {
-		t.Fatalf("unexpected continue-as-new request: %#v", service.continueAsNewRequest)
-	}
+
 }
 
 func TestFlowConfigRejectsWorkerTarget(t *testing.T) {
@@ -499,7 +483,6 @@ func TestFlowMutationsRequireConfirmation(t *testing.T) {
 		{"flow", "start", "flow-1", "--flow-type", "OrderFlow", "--server", address},
 		{"flow", "skip-timer", "flow-1", "--step-type", "Wait", "--condition-id", "timer", "--server", address},
 		{"flow", "update-config", "flow-1", "--config", `{}`, "--server", address},
-		{"flow", "trigger-continue-as-new", "flow-1", "--server", address},
 	} {
 		app := NewApp(bytes.NewReader(nil), &bytes.Buffer{}, &bytes.Buffer{})
 		if err := app.Execute(context.Background(), args); err == nil || ExitCode(err) != 2 {
@@ -508,7 +491,7 @@ func TestFlowMutationsRequireConfirmation(t *testing.T) {
 	}
 	service.mu.Lock()
 	defer service.mu.Unlock()
-	if service.startCalls != 0 || service.skipTimerCalls != 0 || service.updateConfigCalls != 0 || service.continueAsNewCalls != 0 {
+	if service.startCalls != 0 || service.skipTimerCalls != 0 || service.updateConfigCalls != 0 {
 		t.Fatalf("mutation reached service without confirmation: %#v", service)
 	}
 }

cli/internal/command/flow.go

@@ -67,8 +67,6 @@ func (c *flowCommand) Execute(ctx context.Context, args []string, options option
 		return executeWaitStep(c, ctx, args[1:], options)
 	case "update-config":
 		return executeUpdateConfig(c, ctx, args[1:], options)
-	case "trigger-continue-as-new":
-		return executeTriggerContinueAsNew(c, ctx, args[1:], options)
 	case "time-travel":
 		return executeTimeTravel(c, ctx, args[1:], options)
 	case "help", "--help", "-h":
@@ -354,15 +352,15 @@ func resolveRunID(
 	if err != nil {
 		return "", err
 	}
-	resolved := response.GetFlowExecutionId().GetRunId()
+	resolved := response.GetFlowRunId().GetRunId()
 	if resolved == "" {
 		return "", fmt.Errorf("GetFlowSummary response has no run ID")
 	}
 	return resolved, nil
 }
 
 func flowSummary(response *dexpb.GetFlowSummaryResponse) map[string]any {
-	execution := response.GetFlowExecutionId()
+	execution := response.GetFlowRunId()
 	return map[string]any{
 		"flowId": execution.GetFlowId(), "runId": execution.GetRunId(),
 		"firstRunId": response.GetFirstRunId(), "requestId": response.GetRequestId(),
@@ -382,7 +380,7 @@ func protoTimestamp(timestamp *timestamppb.Timestamp) any {
 func (c *flowCommand) printUsage() {
 	fmt.Fprintln(c.stdout, "Usage: dexcli flow <command>")
 	fmt.Fprintln(c.stdout)
-	fmt.Fprintln(c.stdout, "Commands: start, wait, search, summary, state, history, channel-messages, delete-channel-message, inspect, watch, stop, skip-timer, wait-step, update-config, trigger-continue-as-new, time-travel")
+	fmt.Fprintln(c.stdout, "Commands: start, wait, search, summary, state, history, channel-messages, delete-channel-message, inspect, watch, stop, skip-timer, wait-step, update-config, time-travel")
 }
 
 func parseInt32(value string, name string) (int32, error) {

cli/internal/command/flow_client_operations.go

@@ -26,13 +26,11 @@ import (
 )
 
 type flowConfigInput struct {
-	ActiveStepSearchMode         *string         `json:"activeStepSearchMode"`
-	ContinueAsNewThreshold       *int32          `json:"continueAsNewThreshold"`
-	ContinueAsNewPageSizeInBytes *int32          `json:"continueAsNewPageSizeInBytes"`
-	StepDurability               *string         `json:"stepDurability"`
-	WorkerTarget                 json.RawMessage `json:"workerTarget"`
-	AttributeStoreNames          *[]string       `json:"attributeStoreNames"`
-	HistoryWindowEvents          *int32          `json:"historyWindowEvents"`
+	ActiveStepSearchMode *string         `json:"activeStepSearchMode"`
+	StepDurability       *string         `json:"stepDurability"`
+	WorkerTarget         json.RawMessage `json:"workerTarget"`
+	AttributeStoreNames  *[]string       `json:"attributeStoreNames"`
+	HistoryWindowEvents  *int32          `json:"historyWindowEvents"`
 	// Accepted for compatibility and never sent; runs started on server-zd 0.1.23 or later ignore it.
 	DeleteExpiredHistoryObjects *bool `json:"deleteExpiredHistoryObjects"`
 	OffloadHistory              *bool `json:"offloadHistory"`
@@ -241,28 +239,6 @@ func executeUpdateConfig(c *flowCommand, ctx context.Context, args []string, opt
 	})
 }
 
-func executeTriggerContinueAsNew(c *flowCommand, ctx context.Context, args []string, options options) error {
-	flags := newFlagSet("dexcli flow trigger-continue-as-new", c.stderr)
-	yes := flags.Bool("yes", false, "confirm the operation")
-	addCommonFlags(flags, &options)
-	if done, err := parseFlowFlags(flags, args, c.stdout, "dexcli flow trigger-continue-as-new FLOW_ID --yes"); done || err != nil {
-		return err
-	}
-	flowID, err := oneFlowID(flags, "flow trigger-continue-as-new")
-	if err != nil {
-		return err
-	}
-	if !*yes {
-		return newConfirmationError("flow trigger-continue-as-new")
-	}
-	return withFlowService(ctx, options, func(callCtx context.Context, client *flowService) error {
-		if _, callErr := client.service.TriggerContinueAsNew(callCtx, &dexpb.TriggerContinueAsNewRequest{FlowId: flowID}); callErr != nil {
-			return newOperationError("flow trigger-continue-as-new", callErr)
-		}
-		return writeOutput(c.stdout, options.output, map[string]any{"flowId": flowID, "requested": true})
-	})
-}
-
 func startFlowRequest(reader io.Reader, flowID string, flowType string, startStepType string, inputSource string, attributesSource string, configSource string, retryPolicySource string, stepOptionsSource string, timeoutHandlerOptionsSource string, flowTimeout string, timeoutPolicy string, idReusePolicy string, startDelay string, ignoreAlreadyStarted bool, requestID string) (*dexpb.StartFlowRequest, error) {
 	input, err := parseNaturalValue(reader, inputSource)
 	if err != nil {
@@ -473,10 +449,8 @@ func parseFlowConfig(reader io.Reader, source string) (*dexpb.FlowConfig, error)
 		return nil, fmt.Errorf("invalid config JSON: %w", err)
 	}
 	config := &dexpb.FlowConfig{
-		ContinueAsNewThreshold:       input.ContinueAsNewThreshold,
-		ContinueAsNewPageSizeInBytes: input.ContinueAsNewPageSizeInBytes,
-		HistoryWindowEvents:          input.HistoryWindowEvents,
-		OffloadHistory:               input.OffloadHistory,
+		HistoryWindowEvents: input.HistoryWindowEvents,
+		OffloadHistory:      input.OffloadHistory,
 	}
 	if input.AttributeStoreNames != nil {
 		config.AttributeStoreNames = &dexpb.AttributeStoreNames{Names: *input.AttributeStoreNames}

cli/internal/command/flow_operations.go

@@ -163,7 +163,7 @@ func executeInspect(c *flowCommand, ctx context.Context, args []string, options
 		if summaryErr != nil {
 			return newOperationError("flow inspect", summaryErr)
 		}
-		resolvedRunID := summaryResponse.GetFlowExecutionId().GetRunId()
+		resolvedRunID := summaryResponse.GetFlowRunId().GetRunId()
 		history, historyErr := loadHistory(
 			callCtx, client.service, flowID, resolvedRunID, 0, int32(*pageSize), nil,
 			*allHistory, options.noHydrate,
@@ -196,7 +196,6 @@ func executeWatch(c *flowCommand, ctx context.Context, args []string, options op
 	flags := newFlagSet("dexcli flow watch", c.stderr)
 	runID := flags.String("run-id", "", "Flow run ID")
 	fromEventID := flags.Int64("from-event-id", 0, "inclusive internal event cursor")
-	followRuns := flags.Bool("follow-runs", false, "follow Continue-As-New runs")
 	pageSize := flags.Int("page-size", 200, "estimated backend page size")
 	addCommonFlags(flags, &options)
 	if done, err := parseFlowFlags(flags, args, c.stdout, "dexcli flow watch FLOW_ID [flags]"); done || err != nil {
@@ -211,7 +210,7 @@ func executeWatch(c *flowCommand, ctx context.Context, args []string, options op
 	}
 	options.timeout = 0
 	return withStreamingFlowService(ctx, options, func(client *flowService) error {
-		return watchRun(c, ctx, client.service, flowID, *runID, *fromEventID, int32(*pageSize), *followRuns, options)
+		return watchRun(c, ctx, client.service, flowID, *runID, *fromEventID, int32(*pageSize), options)
 	})
 }
 
@@ -223,7 +222,6 @@ func watchRun(
 	runID string,
 	fromEventID int64,
 	pageSize int32,
-	followRuns bool,
 	options options,
 ) error {
 	currentRunID, err := resolveRunID(ctx, client, flowID, runID)
@@ -268,16 +266,6 @@ func watchRun(
 			return newOperationError("flow watch", summaryErr)
 		}
 		if summary.GetFlowStatus() != dexpb.FlowStatus_FLOW_STATUS_RUNNING {
-			if followRuns && summary.GetFlowStatus() == dexpb.FlowStatus_FLOW_STATUS_CONTINUED_AS_NEW {
-				currentRunID, err = resolveRunID(ctx, client, flowID, "")
-				if err != nil {
-					return newOperationError("flow watch", err)
-				}
-				nextEventID = 0
-				nextPageToken = nil
-				lastStateEventID = -1
-				continue
-			}
 			return nil
 		}
 		if nextEventID != lastStateEventID {

docs/design/plan/design-web.md

@@ -3,6 +3,8 @@
 状态：Implementation in progress
 日期：2026-08-05
 
+说明：Dex issue #1072 移除了 run 续接机制。下文涉及该机制的段落记录当时基于 Temporal/Cadence 的设计，不代表当前行为。
+
 ## 1. Phase 1 范围
 
 Phase 1 确定：
@@ -90,7 +92,7 @@ service FlowService {
 ### 4.1 Proto
 
 ```proto
-message FlowExecutionID {
+message FlowRunID {
   string flow_id = 1;
   string run_id = 2;
 }
@@ -102,7 +104,7 @@ message GetFlowSummaryRequest {
 }
 
 message GetFlowSummaryResponse {
-  FlowExecutionID flow_execution_id = 1;
+  FlowRunID flow_run_id = 1;
   string first_run_id = 2;
   string request_id = 3;
   string flow_type = 4;
@@ -217,7 +219,7 @@ message TimeTravelForkHistoryEvent {
 
 ```proto
 message FlowStartedOrContinuedHistoryEvent {
-  FlowExecutionID flow_execution_id = 1;
+  FlowRunID flow_run_id = 1;
   string flow_type = 2;
   FlowConfig flow_config = 3;