From 55b3515d5ea56f8b86f98f79c6994fcdd3e76f41 Mon Sep 17 00:00:00 2001 From: "ark-hand[bot]" Date: Wed, 14 Oct 2026 03:41:10 +0000 Subject: [PATCH] fix(selfhosted): align heartbeat ownership recovery Hand-written change from the internal SDK repository. The original commit message is intentionally not mirrored; see the source repository for details. Sync-Source-Commit: 6f9e2a0878679d3d918318b62659988dee495dfa Hand-Written-Reason: No Ark-APIs provenance marker; treated as a hand-written source commit. Release-Version: 0.11.0 --- arkruntime/environment_work.go | 3 + arkruntime/lib/environments/heartbeat.go | 1 + .../lib/environments/integration_test.go | 3 + arkruntime/model/environment/oas_json_gen.go | 81 ++----------------- .../model/environment/oas_parameters_gen.go | 12 ++- .../model/environment/oas_schemas_gen.go | 45 +---------- arkruntime/model/environment/work_shim.go | 1 + arkruntime/self_hosted_client_test.go | 21 ++--- 8 files changed, 40 insertions(+), 127 deletions(-) diff --git a/arkruntime/environment_work.go b/arkruntime/environment_work.go index 619d4a9..359e7fb 100644 --- a/arkruntime/environment_work.go +++ b/arkruntime/environment_work.go @@ -108,6 +108,9 @@ func (c *Client) HeartbeatWork( if expectedLastHeartbeat, ok := body.ExpectedLastHeartbeat.Get(); ok && expectedLastHeartbeat != "" { q.Set("expected_last_heartbeat", expectedLastHeartbeat) } + if workerID, ok := body.WorkerID.Get(); ok && workerID != "" { + setters = append(setters, WithCustomHeader(environmentWorkWorkerIDHeader, workerID)) + } u := c.fullURL(fmt.Sprintf("%s/%s/work/%s/heartbeat", environmentsPrefix, environment.PathEscape(body.EnvironmentID), diff --git a/arkruntime/lib/environments/heartbeat.go b/arkruntime/lib/environments/heartbeat.go index 003a294..403165a 100644 --- a/arkruntime/lib/environments/heartbeat.go +++ b/arkruntime/lib/environments/heartbeat.go @@ -41,6 +41,7 @@ func (w *EnvironmentWorker) heartbeatLoop(ctx context.Context, work claimedWork, resp, err := api.HeartbeatWork(beatCtx, selfhosted.HeartbeatWorkRequest{ EnvironmentID: work.EnvironmentID, WorkID: work.ID, + WorkerID: environment.NewOptString(w.opts.WorkerID), ExpectedLastHeartbeat: environment.NewOptString(last), DesiredTTLSeconds: environment.NewOptInt64(int64(ttl / time.Second)), }) diff --git a/arkruntime/lib/environments/integration_test.go b/arkruntime/lib/environments/integration_test.go index 39f2017..63eb236 100644 --- a/arkruntime/lib/environments/integration_test.go +++ b/arkruntime/lib/environments/integration_test.go @@ -450,6 +450,9 @@ func TestEnvironmentWorkerHeartbeatLeaseLostOnPreconditionFailed(t *testing.T) { if value, ok := got.ExpectedLastHeartbeat.Get(); !ok || value != selfhosted.ExpectedLastHeartbeatNoHeartbeat { t.Fatalf("expected_last_heartbeat=%+v", got.ExpectedLastHeartbeat) } + if value, ok := got.WorkerID.Get(); !ok || value != "worker_local" { + t.Fatalf("worker_id=%+v", got.WorkerID) + } if value, ok := got.DesiredTTLSeconds.Get(); !ok || value != int64(heartbeatDefault/time.Second) { t.Fatalf("desired_ttl_seconds=%+v", got.DesiredTTLSeconds) } diff --git a/arkruntime/model/environment/oas_json_gen.go b/arkruntime/model/environment/oas_json_gen.go index 6761fa8..afcc108 100644 --- a/arkruntime/model/environment/oas_json_gen.go +++ b/arkruntime/model/environment/oas_json_gen.go @@ -384,12 +384,6 @@ func (s *EnvConfig) encodeFields(e *jx.Encoder) { s.SetupScript.Encode(e) } } - { - if s.ActiveRecovery.Set { - e.FieldStart("active_recovery") - s.ActiveRecovery.Encode(e) - } - } { if s.Tos.Set { e.FieldStart("tos") @@ -398,14 +392,13 @@ func (s *EnvConfig) encodeFields(e *jx.Encoder) { } } -var jsonFieldsNameOfEnvConfig = [7]string{ +var jsonFieldsNameOfEnvConfig = [6]string{ 0: "type", 1: "networking", 2: "packages", 3: "env", 4: "setup_script", - 5: "active_recovery", - 6: "tos", + 5: "tos", } // Decode decodes EnvConfig from json. @@ -467,16 +460,6 @@ func (s *EnvConfig) Decode(d *jx.Decoder) error { }(); err != nil { return errors.Wrap(err, "decode field \"setup_script\"") } - case "active_recovery": - if err := func() error { - s.ActiveRecovery.Reset() - if err := s.ActiveRecovery.Decode(d); err != nil { - return err - } - return nil - }(); err != nil { - return errors.Wrap(err, "decode field \"active_recovery\"") - } case "tos": if err := func() error { s.Tos.Reset() @@ -1756,41 +1739,6 @@ func (s *OptEnvironmentScope) UnmarshalJSON(data []byte) error { return s.Decode(d) } -// Encode encodes int32 as json. -func (o OptInt32) Encode(e *jx.Encoder) { - if !o.Set { - return - } - e.Int32(int32(o.Value)) -} - -// Decode decodes int32 from json. -func (o *OptInt32) Decode(d *jx.Decoder) error { - if o == nil { - return errors.New("invalid: unable to decode OptInt32 to nil") - } - o.Set = true - v, err := d.Int32() - if err != nil { - return err - } - o.Value = int32(v) - return nil -} - -// MarshalJSON implements stdjson.Marshaler. -func (s OptInt32) MarshalJSON() ([]byte, error) { - e := jx.Encoder{} - s.Encode(&e) - return e.Bytes(), nil -} - -// UnmarshalJSON implements stdjson.Unmarshaler. -func (s *OptInt32) UnmarshalJSON(data []byte) error { - d := jx.DecodeBytes(data) - return s.Decode(d) -} - // Encode encodes NetworkingConfig as json. func (o OptNetworkingConfig) Encode(e *jx.Encoder) { if !o.Set { @@ -2993,19 +2941,13 @@ func (s *WorkItem) encodeFields(e *jx.Encoder) { s.StopReason.Encode(e) } } - { - if s.RecoveryCount.Set { - e.FieldStart("recovery_count") - s.RecoveryCount.Encode(e) - } - } { e.FieldStart("type") s.Type.Encode(e) } } -var jsonFieldsNameOfWorkItem = [15]string{ +var jsonFieldsNameOfWorkItem = [14]string{ 0: "id", 1: "acknowledged_at", 2: "created_at", @@ -3019,8 +2961,7 @@ var jsonFieldsNameOfWorkItem = [15]string{ 10: "stop_requested_at", 11: "stopped_at", 12: "stop_reason", - 13: "recovery_count", - 14: "type", + 13: "type", } // Decode decodes WorkItem from json. @@ -3175,18 +3116,8 @@ func (s *WorkItem) Decode(d *jx.Decoder) error { }(); err != nil { return errors.Wrap(err, "decode field \"stop_reason\"") } - case "recovery_count": - if err := func() error { - s.RecoveryCount.Reset() - if err := s.RecoveryCount.Decode(d); err != nil { - return err - } - return nil - }(); err != nil { - return errors.Wrap(err, "decode field \"recovery_count\"") - } case "type": - requiredBitSet[1] |= 1 << 6 + requiredBitSet[1] |= 1 << 5 if err := func() error { if err := s.Type.Decode(d); err != nil { return err @@ -3206,7 +3137,7 @@ func (s *WorkItem) Decode(d *jx.Decoder) error { var failures []validate.FieldError for i, mask := range [2]uint8{ 0b00011101, - 0b01000010, + 0b00100010, } { if result := (requiredBitSet[i] & mask) ^ mask; result != 0 { // Mask only required fields and check equality to mask using XOR. diff --git a/arkruntime/model/environment/oas_parameters_gen.go b/arkruntime/model/environment/oas_parameters_gen.go index 98c60bf..ccdd33c 100644 --- a/arkruntime/model/environment/oas_parameters_gen.go +++ b/arkruntime/model/environment/oas_parameters_gen.go @@ -9,7 +9,9 @@ package environment type EnvironmentWorkAckParams struct { EnvironmentId string WorkId string - // Worker 实例 ID，用于控制面记录 work 归属和排障。. + // Worker 实例 ID。Poll/Ack 用于记录 work owner；Heartbeat 携带非空值时必须与 + // Ack 保存的 owner 精确一致，否则返回 412。匹配 owner 的 Heartbeat 可将 + // `stopped + lease_expired` 直接恢复为 `active`。. ArkWorkerID OptString `json:",omitempty,omitzero"` } @@ -21,6 +23,10 @@ type EnvironmentWorkHeartbeatParams struct { DesiredTTLSeconds OptInt64 `json:",omitempty,omitzero" query:"desired_ttl_seconds"` // Worker 上一次看到的 last_heartbeat，用于并发校验。. ExpectedLastHeartbeat OptString `json:",omitempty,omitzero" query:"expected_last_heartbeat"` + // Worker 实例 ID。Poll/Ack 用于记录 work owner；Heartbeat 携带非空值时必须与 + // Ack 保存的 owner 精确一致，否则返回 412。匹配 owner 的 Heartbeat 可将 + // `stopped + lease_expired` 直接恢复为 `active`。. + ArkWorkerID OptString `json:",omitempty,omitzero"` } // EnvironmentWorkPollParams is parameters of EnvironmentWork_poll operation. @@ -30,7 +36,9 @@ type EnvironmentWorkPollParams struct { BlockMs OptInt64 `json:",omitempty,omitzero" query:"block_ms"` // 允许控制面回收超过指定时长未 heartbeat 的 work，单位毫秒。. ReclaimOlderThanMs OptInt64 `json:",omitempty,omitzero" query:"reclaim_older_than_ms"` - // Worker 实例 ID，用于控制面记录 work 归属和排障。. + // Worker 实例 ID。Poll/Ack 用于记录 work owner；Heartbeat 携带非空值时必须与 + // Ack 保存的 owner 精确一致，否则返回 412。匹配 owner 的 Heartbeat 可将 + // `stopped + lease_expired` 直接恢复为 `active`。. ArkWorkerID OptString `json:",omitempty,omitzero"` } diff --git a/arkruntime/model/environment/oas_schemas_gen.go b/arkruntime/model/environment/oas_schemas_gen.go index 3048964..339918f 100644 --- a/arkruntime/model/environment/oas_schemas_gen.go +++ b/arkruntime/model/environment/oas_schemas_gen.go @@ -128,15 +128,6 @@ type EnvConfig struct { Env OptEnvConfigEnv `json:"env"` // 沙箱启动阶段执行的初始化脚本。. SetupScript OptString `json:"setup_script"` - // 是否启用 Work 积极恢复；仅 self_hosted Environment 支持，省略时等同于 - // false。该配置 - // 仅允许在 Environment 级设置，不允许通过 Session override/upgrade - // 覆盖。开启后，平台在 - // Work 已 stopped、仍有未闭合 agent.tool_use 且 recovery_count 少于 5 次时将 Work - // 重新排队， - // 不按 stop_reason 排除。只有 Worker 上报 completed 且全部 tool_use - // 已闭合后才会清零计数。. - ActiveRecovery OptBool `json:"active_recovery"` // Environment outputs 的 TOS 存储配置。. Tos OptTosConfig `json:"tos"` } @@ -166,11 +157,6 @@ func (s *EnvConfig) GetSetupScript() OptString { return s.SetupScript } -// GetActiveRecovery returns the value of ActiveRecovery. -func (s *EnvConfig) GetActiveRecovery() OptBool { - return s.ActiveRecovery -} - // GetTos returns the value of Tos. func (s *EnvConfig) GetTos() OptTosConfig { return s.Tos @@ -201,11 +187,6 @@ func (s *EnvConfig) SetSetupScript(val OptString) { s.SetupScript = val } -// SetActiveRecovery sets the value of ActiveRecovery. -func (s *EnvConfig) SetActiveRecovery(val OptBool) { - s.ActiveRecovery = val -} - // SetTos sets the value of Tos. func (s *EnvConfig) SetTos(val OptTosConfig) { s.Tos = val @@ -1580,10 +1561,8 @@ type StopWorkBody struct { // `worker_abnormal`；用户或管理员操作分别传 // `user_cancelled` 或 `admin_stopped`，无法归类时传 `others`。`lease_expired` // 仅由平台在租约 - // 过期时写入，不应通过 StopWork 上报。该字段用于停止诊断，并在 Worker - // 上报 `completed` - // 且全部 tool_use - // 已闭合时作为清零恢复次数的信号；平台不会按停止原因排除积极恢复。. + // 过期时写入，不应通过 StopWork + // 上报。该字段仅用于停止诊断；强停不会自动重新排队。. Reason OptWorkStopReason `json:"reason"` } @@ -1804,15 +1783,9 @@ type WorkItem struct { // Work 停止时间，RFC 3339。. StoppedAt OptString `json:"stopped_at"` // 最近一次停止原因。首次停止前省略；state 为 stopping / stopped - // 时返回本次停止原因； - // 自动恢复到 queued / starting / active - // 后仍保留上一次停止原因，当前生命周期状态以 - // `state` 为准。. + // 时返回本次停止原因。 + // 原 owner 通过 Heartbeat 将 `stopped + lease_expired` 恢复为 active 时清除此字段。. StopReason OptWorkStopReason `json:"stop_reason"` - // 自上次成功完成且全部 tool_use 已闭合后，平台已自动将本 Work - // 重新排队的次数； - // 首次执行或字段省略时按 0 处理。手动 Ensure 不会清零该计数。. - RecoveryCount OptInt32 `json:"recovery_count"` // 对象类型，固定为 `work`。. Type WorkItemType `json:"type"` } @@ -1882,11 +1855,6 @@ func (s *WorkItem) GetStopReason() OptWorkStopReason { return s.StopReason } -// GetRecoveryCount returns the value of RecoveryCount. -func (s *WorkItem) GetRecoveryCount() OptInt32 { - return s.RecoveryCount -} - // GetType returns the value of Type. func (s *WorkItem) GetType() WorkItemType { return s.Type @@ -1957,11 +1925,6 @@ func (s *WorkItem) SetStopReason(val OptWorkStopReason) { s.StopReason = val } -// SetRecoveryCount sets the value of RecoveryCount. -func (s *WorkItem) SetRecoveryCount(val OptInt32) { - s.RecoveryCount = val -} - // SetType sets the value of Type. func (s *WorkItem) SetType(val WorkItemType) { s.Type = val diff --git a/arkruntime/model/environment/work_shim.go b/arkruntime/model/environment/work_shim.go index 8e72fa9..1981240 100644 --- a/arkruntime/model/environment/work_shim.go +++ b/arkruntime/model/environment/work_shim.go @@ -45,6 +45,7 @@ type AckWorkRequest struct { type HeartbeatWorkRequest struct { EnvironmentID string `json:"environment_id"` WorkID string `json:"work_id"` + WorkerID OptString `json:"worker_id,omitempty"` ExpectedLastHeartbeat OptString `json:"expected_last_heartbeat,omitempty"` DesiredTTLSeconds OptInt64 `json:"desired_ttl_seconds,omitempty"` } diff --git a/arkruntime/self_hosted_client_test.go b/arkruntime/self_hosted_client_test.go index 5b06e13..f7b3f3b 100644 --- a/arkruntime/self_hosted_client_test.go +++ b/arkruntime/self_hosted_client_test.go @@ -17,7 +17,10 @@ import ( "github.com/volcengine/ark-runtime-go/arkruntime/model/session" ) -const testBearerToken = "Bearer test-api-key" +const ( + testBearerToken = "Bearer test-api-key" + testWorkerID = "worker-1" +) func TestEnvironmentWorkRequests(t *testing.T) { var seen []string @@ -33,7 +36,7 @@ func TestEnvironmentWorkRequests(t *testing.T) { switch r.Method + " " + r.URL.Path { case "GET /environments/env-1/work/poll": - if got := r.Header.Get(environmentWorkWorkerIDHeader); got != "worker-1" { + if got := r.Header.Get(environmentWorkWorkerIDHeader); got != testWorkerID { t.Fatalf("Ark-Worker-ID = %q", got) } if got := r.URL.Query().Get("block_ms"); got != "999" { @@ -53,7 +56,6 @@ func TestEnvironmentWorkRequests(t *testing.T) { "latest_heartbeat_at": "2026-08-10T00:00:00Z", "state": "queued", "stop_reason": "worker_abnormal", - "recovery_count": 5, "type": "work", } if err := json.NewEncoder(w).Encode(pollResponse); err != nil { @@ -61,12 +63,15 @@ func TestEnvironmentWorkRequests(t *testing.T) { } case "POST /environments/env-1/work/work-1/ack": assertNoBody(t, r) - if got := r.Header.Get(environmentWorkWorkerIDHeader); got != "worker-1" { + if got := r.Header.Get(environmentWorkWorkerIDHeader); got != testWorkerID { t.Fatalf("Ark-Worker-ID = %q", got) } _, _ = w.Write([]byte(`{"id":"work-1","created_at":"2026-08-10T00:00:00Z","data":{"id":"sess-1","type":"session"},"environment_id":"env-1","state":"starting","type":"work"}`)) case "POST /environments/env-1/work/work-1/heartbeat": assertNoBody(t, r) + if got := r.Header.Get(environmentWorkWorkerIDHeader); got != testWorkerID { + t.Fatalf("Ark-Worker-ID = %q", got) + } if got := r.URL.Query().Get("desired_ttl_seconds"); got != "30" { t.Fatalf("desired_ttl_seconds = %q", got) } @@ -95,7 +100,7 @@ func TestEnvironmentWorkRequests(t *testing.T) { work, err := client.PollWork(ctx, &environment.PollWorkRequest{ EnvironmentID: "env-1", - WorkerID: "worker-1", + WorkerID: testWorkerID, BlockMS: 999, }) if err != nil { @@ -107,19 +112,17 @@ func TestEnvironmentWorkRequests(t *testing.T) { if got := work.LatestHeartbeatValue(); got != "2026-08-10T00:00:00Z" { t.Fatalf("PollWork().LatestHeartbeatValue() = %q", got) } - if got, ok := work.RecoveryCount.Get(); !ok || got != 5 { - t.Fatalf("PollWork().RecoveryCount = %d, set=%v", got, ok) - } if err := client.AckWork(ctx, &environment.AckWorkRequest{ EnvironmentID: "env-1", WorkID: "work-1", - WorkerID: environment.NewOptString("worker-1"), + WorkerID: environment.NewOptString(testWorkerID), }); err != nil { t.Fatalf("AckWork() error = %v", err) } heartbeat, err := client.HeartbeatWork(ctx, &environment.HeartbeatWorkRequest{ EnvironmentID: "env-1", WorkID: "work-1", + WorkerID: environment.NewOptString(testWorkerID), DesiredTTLSeconds: environment.NewOptInt64(30), ExpectedLastHeartbeat: environment.NewOptString("2026-08-10T00:00:00Z"), })