Compact Flow-scoped Blob refs and make ASYNC input snapshots opt in

cli/internal/command/command_integ_test.go

@@ -32,6 +32,7 @@ type testFlowService struct {
 
 	mu                   sync.Mutex
 	loadCalls            int
+	loadRequest          *dexpb.LoadBlobsRequest
 	startCalls           int
 	stopCalls            int
 	streamWriteCalls     int
@@ -85,11 +86,12 @@ func (s *testFlowService) SearchFlows(
 }
 
 func (s *testFlowService) LoadBlobs(
-	context.Context,
-	*dexpb.LoadBlobsRequest,
+	_ context.Context,
+	request *dexpb.LoadBlobsRequest,
 ) (*dexpb.LoadBlobsResponse, error) {
 	s.mu.Lock()
 	s.loadCalls++
+	s.loadRequest = request
 	s.mu.Unlock()
 	return &dexpb.LoadBlobsResponse{Values: map[string]*dexpb.Value{
 		"blob-1": {Kind: &dexpb.Value_StringValue{StringValue: "hydrated"}},
@@ -283,6 +285,10 @@ func TestSearchHydratesByDefaultAndCanReturnReferences(t *testing.T) {
 	if service.loadCalls != 1 {
 		t.Fatalf("expected one hydration call, got %d", service.loadCalls)
 	}
+	if len(service.loadRequest.GetEntries()) != 1 ||
+		service.loadRequest.GetEntries()[0].GetFlowId() != "flow-1" {
+		t.Fatalf("unexpected hydration request: %#v", service.loadRequest)
+	}
 }
 
 func TestMutationRequiresYesBeforeSendingRequest(t *testing.T) {

cli/internal/command/flow.go

@@ -101,7 +101,7 @@ func (c *flowCommand) channelMessages(ctx context.Context, args []string, option
 		if callErr != nil {
 			return newOperationError("flow channel-messages", callErr)
 		}
-		mapped, warnings, mapErr := naturalMessage(callCtx, client.service, response, options.noHydrate)
+		mapped, warnings, mapErr := naturalMessage(callCtx, client.service, flowID, response, options.noHydrate)
 		if mapErr != nil {
 			return newOperationError("flow channel-messages", mapErr)
 		}
@@ -177,7 +177,7 @@ func (c *flowCommand) search(ctx context.Context, args []string, options options
 				return newOperationError("flow search", err)
 			}
 			for _, entry := range response.GetFlowRuns() {
-				mapped, warnings, mapErr := naturalMessage(callCtx, client.service, entry, options.noHydrate)
+				mapped, warnings, mapErr := naturalMessage(callCtx, client.service, entry.GetFlowId(), entry, options.noHydrate)
 				if mapErr != nil {
 					return newOperationError("flow search", mapErr)
 				}
@@ -242,7 +242,7 @@ func (c *flowCommand) state(ctx context.Context, args []string, options options)
 		if callErr != nil {
 			return newOperationError("flow state", callErr)
 		}
-		mapped, warnings, mapErr := naturalMessage(callCtx, client.service, response, options.noHydrate)
+		mapped, warnings, mapErr := naturalMessage(callCtx, client.service, flowID, response, options.noHydrate)
 		if mapErr != nil {
 			return newOperationError("flow state", mapErr)
 		}

cli/internal/command/flow_client_operations.go

@@ -126,7 +126,7 @@ func executeWait(c *flowCommand, ctx context.Context, args []string, options opt
 		if callErr != nil {
 			return newOperationError("flow wait", callErr)
 		}
-		mapped, warnings, mapErr := naturalMessage(callCtx, client.service, response, options.noHydrate)
+		mapped, warnings, mapErr := naturalMessage(callCtx, client.service, flowID, response, options.noHydrate)
 		if mapErr != nil {
 			return newOperationError("flow wait", mapErr)
 		}

cli/internal/command/flow_operations.go

@@ -44,7 +44,7 @@ func loadHistory(
 			return nil, err
 		}
 		for _, event := range response.GetEvents() {
-			mapped, valueWarnings, mapErr := naturalMessage(ctx, client, event, noHydrate)
+			mapped, valueWarnings, mapErr := naturalMessage(ctx, client, flowID, event, noHydrate)
 			if mapErr != nil {
 				return nil, mapErr
 			}
@@ -179,7 +179,7 @@ func executeInspect(c *flowCommand, ctx context.Context, args []string, options
 			if stateErr != nil {
 				return newOperationError("flow inspect", stateErr)
 			}
-			state, warnings, mapErr := naturalMessage(callCtx, client.service, stateResponse, options.noHydrate)
+			state, warnings, mapErr := naturalMessage(callCtx, client.service, flowID, stateResponse, options.noHydrate)
 			if mapErr != nil {
 				return newOperationError("flow inspect", mapErr)
 			}
@@ -242,7 +242,7 @@ func watchRun(
 			return newOperationError("flow watch", pageErr)
 		}
 		for _, event := range page.GetEvents() {
-			mapped, warnings, mapErr := naturalMessage(ctx, client, event, options.noHydrate)
+			mapped, warnings, mapErr := naturalMessage(ctx, client, flowID, event, options.noHydrate)
 			if mapErr != nil {
 				return newOperationError("flow watch", mapErr)
 			}
@@ -287,7 +287,7 @@ func watchRun(
 			if stateErr != nil {
 				return newOperationError("flow watch", stateErr)
 			}
-			state, warnings, mapErr := naturalMessage(ctx, client, stateResponse, options.noHydrate)
+			state, warnings, mapErr := naturalMessage(ctx, client, flowID, stateResponse, options.noHydrate)
 			if mapErr != nil {
 				return newOperationError("flow watch", mapErr)
 			}

cli/internal/command/values.go

@@ -28,6 +28,7 @@ type blobReference struct {
 func naturalMessage(
 	ctx context.Context,
 	client dexpb.FlowServiceClient,
+	flowID string,
 	message proto.Message,
 	noHydrate bool,
 ) (map[string]any, []string, error) {
@@ -40,7 +41,7 @@ func naturalMessage(
 	if noHydrate || len(references) == 0 {
 		return naturalValue(raw, nil, false).(map[string]any), nil, nil
 	}
-	replacements, warnings := hydrateBlobReferences(ctx, client, references)
+	replacements, warnings := hydrateBlobReferences(ctx, client, flowID, references)
 	return naturalValue(raw, replacements, true).(map[string]any), warnings, nil
 }
 
@@ -76,9 +77,10 @@ func collectBlobReferences(value any, references map[string]blobReference) {
 func hydrateBlobReferences(
 	ctx context.Context,
 	client dexpb.FlowServiceClient,
+	flowID string,
 	references map[string]blobReference,
 ) (map[string]any, []string) {
-	values := make([]*dexpb.Value, 0, len(references))
+	entries := make([]*dexpb.LoadBlobRequestEntry, 0, len(references))
 	for _, reference := range references {
 		value := &dexpb.Value{}
 		if reference.kind == "string" {
@@ -90,9 +92,9 @@ func hydrateBlobReferences(
 				InternalBlobIdForObjValue: reference.id,
 			}
 		}
-		values = append(values, value)
+		entries = append(entries, &dexpb.LoadBlobRequestEntry{FlowId: flowID, BlobValue: value})
 	}
-	response, err := client.LoadBlobs(ctx, &dexpb.LoadBlobsRequest{Values: values})
+	response, err := client.LoadBlobs(ctx, &dexpb.LoadBlobsRequest{Entries: entries})
 	if err != nil {
 		return nil, []string{fmt.Sprintf("stored values unavailable: %v", err)}
 	}

docs/content/production/server-operations.mdx

@@ -91,6 +91,84 @@ Use separate credentials and least-privilege policies for each environment.
 Monitor write and read failures, then test a recovery procedure that can still
 read existing objects after a credential rotation.
 
+Dex keeps payloads through 100 bytes inline by default. A 101-byte string or
+encoded object is offloaded. Set
+**blobStore.thresholdInBytes** to override the default. The comparison uses only
+the payload size, not the resulting reference size.
+
+~~~yaml
+blobStore:
+  thresholdInBytes: 100
+  objectIdLength: 10
+  asyncStepInputSnapshotsEnabled: false
+  supportedStorages:
+    - status: active
+      storageId: p1
+      storageType: s3
+~~~
+
+Keep **storageId** short because every offloaded value stores it in durable
+history. For example, use **p1** instead of **production1** when the shorter name
+is still operationally clear.
+
+Blob references are opaque application values. Their wire forms are:
+
+~~~text
+String: <storageId>|<YYMMDD>/<objectId>
+Object: <storageId>|<YYMMDD>/<objectId>
+
+Example: p1|260913/ab3de7kp2x
+~~~
+
+The Value arm distinguishes String and Object references. An Object Blob stores
+the complete **EncodedObject**, including its **json**, **raw**, or custom
+encoding. The reference keeps the storage ID and UTC write date, but omits the
+Flow ID and encoding. Dex obtains the owning Flow ID from trusted request or
+Worker context. The physical object path is:
+
+~~~text
+<namespace>/<YYMMDD>$<base64url(flowId)>/<objectId>
+~~~
+
+The two-digit year represents 2000 through 2099.
+
+Internal references belong to exactly one Flow. Dex rehydrates and rewrites a
+reference before it crosses a Flow boundary. The destination receives an
+inline value or a new object under its own Flow prefix. Deleting the source
+Flow therefore cannot corrupt the destination Flow. Public APIs reject
+client-supplied internal references.
+
+Object IDs use deterministic lowercase Base36 characters. Retries with the
+same invocation ID and stored bytes overwrite the same object instead of
+creating duplicates. Object stored bytes include the encoding. Dex does not
+perform a read-before-write, conditional create, or collision retry.
+**blobStore.objectIdLength** defaults to 10. Zero selects that default, negative
+values are invalid, and positive values are accepted without a protocol-defined
+range. Every Server writing the same Blob Store namespace must use
+the same value, and the setting must not change while those Servers are
+running. Readers accept any non-empty lowercase Base36 ID, so a deployment can
+change the configured length during a coordinated restart.
+
+Ten Base36 characters provide about 51.7 bits. Within one Flow and one UTC day,
+the approximate probability of at least one collision is 1.4 × 10⁻⁸ for 10,000
+objects, 1.4 × 10⁻⁶ for 100,000 objects, and 1.4 × 10⁻⁴ for 1,000,000 objects.
+Use 10 for ordinary deployments. Use 12 or 16 for high-throughput Flows. A
+length of 50 can represent the full SHA-256 value. Longer values add leading
+zeros but no additional entropy.
+
+Cleanup keeps the UTC date at the start of each Flow prefix. It lists prefixes
+in lexical pages, describes the encoded Flow and optional Run, and deletes the
+whole prefix only after the execution no longer exists. Continue listing with
+the returned continuation token; no manifest is involved.
+
+Async Step input snapshots are disabled by default. Set
+**blobStore.asyncStepInputSnapshotsEnabled** to true only when semantic history
+must retain the exact inputs sent to successful ASYNC local Step methods. The
+snapshot is independent of the payload offload threshold and is not required
+for Flow execution, retry, or recovery. When disabled, ASYNC local completion
+events report their input as unavailable. SYNC methods and ASYNC methods that
+fall back to a regular Activity continue to obtain input from backend history.
+
 The optional **blobStore.blobCache** caches S3-backed Attribute objects. A
 non-empty **directory** enables it and must belong to one Server process. Its
 default budget is 1 GiB. Oversized or rejected objects bypass the cache and

docs/design/deterministic-blob-identity.md

@@ -18,38 +18,59 @@ For `StartFlow`, the server stores the request ID in workflow memo. When
 `ignore_already_started_error` is enabled, an AlreadyStarted error is ignored
 only when the running workflow has the same request ID.
 
-`FlowAlreadyStartedOptions.request_id` is removed and its field number and name
-are reserved.
+`FlowAlreadyStartedOptions.request_id` is removed without a compatibility field.
 
-## Blob UUID
+## Blob object ID
 
-Blob object names use deterministic UUIDv8:
+Blob object names use a configurable deterministic lowercase Base36 ID:
 
 ```text
 digest = SHA-256(lengthPrefixed(
-  "dex-blob-v1",
+  "dex-blob-v2",
   invocationID,
-  payload,
+  storedBytes,
 ))
-blobUUID = UUIDv8(digest[0:16])
+objectID = fixedWidthBase36(digest mod 36^objectIdLength)
 ```
 
 Each component is prefixed with its unsigned 64-bit big-endian byte length. The
-first 16 digest bytes are copied before setting the UUIDv8 version and RFC
-variant bits.
+alphabet is `0123456789abcdefghijklmnopqrstuvwxyz`. The configured length defaults
+to 10. Zero selects the default, negative values are invalid, and positive
+values have no protocol-defined range. All Servers writing the same namespace
+use the same immutable value. Lengths above 50 add leading zeros without adding
+SHA-256 entropy.
 
-The complete path remains:
+String blobs store their UTF-8 bytes. Object blobs store a deterministic protobuf
+serialization of the complete `EncodedObject`, including its encoding and
+payload. The offload threshold still compares the original payload length.
+
+Durable History stores a compact locator without the Flow ID or encoding:
 
 ```text
-yyyyMMdd$flowID/<blobUUID>
+<storageId>|<yyMMdd>/<objectID>
+```
+
+The Server combines the locator with the contextual Flow ID:
+
+```text
+<namespace>/<yyMMdd>$<base64url(flowID)>/<objectID>
 ```
 
 Activity writes use the workflow run ID plus activity ID as `invocationID`.
 External API writes use the caller-provided request ID. Activity attempt numbers
 are excluded.
 
+Normal writes overwrite an existing key. There is no conditional create,
+read-before-write, collision retry, or fallback key. This accepts the
+probabilistic collision risk of the configured truncated hash.
+
+Internal Blob references belong to one Flow. Before a reference crosses a Flow
+boundary, the Server reads it with the source Flow ID and writes any still-large
+value under the destination Flow ID. This makes cleanup of the source Flow safe.
+
 The date prefix uses the server's UTC date when the object is written. A retry
-crossing a UTC date boundary can create another path.
+crossing a UTC date boundary can create another path. Its two-digit year covers
+2000 through 2099.
 
 S3 bucket versioning must remain disabled. Otherwise, repeated writes to one
 deterministic key retain hidden object versions.

docs/design/plan/design-web.md

@@ -370,13 +370,18 @@ Web 只消费统一的 `input/output/context`，不根据 durability 选择额
 | 执行路径 | Input 来源 | Context/options 来源 |
 |---|---|---|
 | SYNC regular Activity | ActivityTaskScheduled input | scheduled event metadata 和 input context |
-| ASYNC local success | run-scoped async input snapshot | LocalActivity marker 与 async input snapshot |
+| ASYNC local success with snapshots enabled | run-scoped async input snapshot | LocalActivity marker 与 async input snapshot |
+| ASYNC local success with snapshots disabled | unavailable | LocalActivity marker metadata |
 | ASYNC local failure + regular fallback | fallback ActivityTaskScheduled input | scheduled event metadata；durability 仍为 ASYNC |
 | ASYNC local failure + budget exhausted | unavailable | LocalActivity failure marker metadata |
 
-local snapshot 不存在、external storage 未启用或数据已清理时，server 返回
+`blobStore.asyncStepInputSnapshotsEnabled` 默认关闭。只有明确开启后，成功的 ASYNC
+local activity 才保存 snapshot。配置关闭、local snapshot 不存在、external storage 未启用
+或数据已清理时，server 返回
 `input.unavailable=true`。这只代表 step method input snapshot 不可恢复，不代表其中某个
-独立 Value blob 加载失败。Web 不显示 page-level data warning；terminal ASYNC failure
+独立 Value blob 加载失败。Web 将它显示为整个 step method input snapshot unavailable，
+将默认关闭及 `blobStore.asyncStepInputSnapshotsEnabled` 列为首要可能原因，并明确区分
+单个 Value blob load failure。Web 不显示 page-level data warning；terminal ASYNC failure
 说明 short retry budget 可在 sync fallback 前耗尽，因此没有记录 invocation
 input snapshot，并引导用户沿 Timeline source link 回看调度来源。
 
@@ -474,7 +479,7 @@ message InternalLocalStepActivityFailure {
 
 - `InternalLocalActivityInput` 是 workflow provider 只给 local activity 的第二个参数，
   用于携带当前 run start time 和 method options；
-- `InternalAsyncStepInputSnapshot` 是成功 local activity 写入 external storage 的 protobuf，
+- `InternalAsyncStepInputSnapshot` 是启用 snapshot 后由成功 local activity 写入 external storage 的 protobuf，
   保存准确发送给 worker 的 request 和 method options；
 - regular failure 使用单一 `InternalActivityError` detail；local failure 使用单一
   `InternalLocalStepActivityFailure` detail，并在其中嵌套 `activity_error`；
@@ -730,12 +735,12 @@ Phase 2 使用 `server/integ/`：
 - Temporal/Cadence × SYNC/ASYNC：WaitFor/Execute 显示调用时 step input、attributes 和 condition results。
 - SYNC scheduled input 和 ASYNC snapshot 都映射为完全相同的 `input/output/context` shape。
 - regular Activity input proto 保持不变；第二个 activity argument 为 null 时 Temporal/Cadence 都能解码。
-- ASYNC local success 保存 `InternalAsyncStepInputSnapshot`；marker 中不增加完整 request。
+- 开启 `blobStore.asyncStepInputSnapshotsEnabled` 后，ASYNC local success 保存 `InternalAsyncStepInputSnapshot`；marker 中不增加完整 request。
 - method options：SYNC 从 scheduled metadata 转换；ASYNC success 从 snapshot 恢复，fallback 从 local failure metadata 恢复。
 - channel values、多个 timers、ANY/ALL results 从保存的 worker request 精确恢复。
 - local failure fallback 使用 regular Activity history request，且不暴露 local failure。
 - sync 和 async regular retry 只返回最近一次 failure；local failure 在 fallback 期间不暴露。
-- local retry budget 耗尽且没有 fallback 时，以及关闭存储或清理后缺失 async snapshot 时，返回 `input.unavailable=true`。
+- local retry budget 耗尽且没有 fallback 时，以及 snapshot 配置关闭、存储关闭或清理后缺失 async snapshot 时，返回 `input.unavailable=true`。
 - local filesystem storage 覆盖 string/object blob、run-level cleanup 和安全路径。
 
 Web Go integration：

docs/design/plan/go-sdk-rewrite.md

@@ -134,19 +134,19 @@ remains internal.
 | unsigned integers up to `math.MaxInt64` | `int_value` |
 | float32 and float64 | `double_value` |
 | bool and named bools | `bool_value` |
-| `[]byte` and named byte slices | `obj_value`, encoding `"rawbytes"` |
+| ordinary nil and typed nil | `null_value` |
+| `[]byte` and named byte slices | `obj_value`, encoding `"raw"` |
 | all other JSON-compatible values | `obj_value`, encoding `"json"` |
 
 Strings with invalid UTF-8 return an encoding error; arbitrary binary data uses
 `[]byte`. Raw-byte object payloads contain the bytes directly without JSON or
 base64 encoding. Structs, maps, other slices, arrays, and non-indexed
-`time.Time` use JSON. Ordinary nil and typed nil encode as a JSON null object.
-The proto null arm is reserved for attribute deletion.
+`time.Time` use JSON. The proto null arm represents top-level null. Within an
+Attribute write, it deletes the Attribute.
 
 `Value.Decode` requires a non-nil pointer. It rejects overflow, incompatible
-targets, malformed JSON, unknown object encodings, deletion markers, and blob
-arms that have not passed through hydration. Dynamic failures return errors and
-never panic.
+targets, malformed JSON, unknown object encodings, and blob arms that have not
+passed through hydration. Dynamic failures return errors and never panic.
 
 ### Indexed attributes
 

docs/design/rust-sdk-user-interface.md

@@ -215,7 +215,7 @@ fn execute(&self, context: &mut Context, input: Import) -> HandlerResult<StepDec
 }
 ```
 
-`record_heartbeat()` emits no Value and clears backend heartbeat details. An encoded JSON null is a
+`record_heartbeat()` emits no Value and clears backend heartbeat details. A null Value is a
 present Value, so decoding `Option<T>` returns outer `Some` with inner `None`. A Stream frame is an
 implicit backend heartbeat that reuses the last explicit Worker heartbeat value, including its
 absence. Local activities ignore heartbeat details but still forward Stream frames.