From d709af800e6e5c4aa288259430a429a4805059a9 Mon Sep 17 00:00:00 2001 From: "ark-hand[bot]" Date: Thu, 8 Oct 2026 09:31:15 +0000 Subject: [PATCH] feat(selfhosted): report work stop reasons 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: 289dc712cf78ba19bb37251ecda6445100e5b26b Hand-Written-Reason: No Ark-APIs provenance marker; treated as a hand-written source commit. Release-Version: 0.10.0 --- arkruntime/environment_work.go | 19 +- .../lib/environments/integration_test.go | 89 +++++++- arkruntime/lib/environments/poller.go | 1 + arkruntime/lib/environments/poller_test.go | 22 ++ arkruntime/lib/environments/worker.go | 23 +- arkruntime/model/environment/oas_json_gen.go | 198 +++++++++++++++++- .../model/environment/oas_schemas_gen.go | 186 ++++++++++++++++ .../model/environment/oas_validators_gen.go | 67 ++++++ arkruntime/model/environment/work_shim.go | 2 + arkruntime/self_hosted_client_test.go | 30 ++- 10 files changed, 621 insertions(+), 16 deletions(-) diff --git a/arkruntime/environment_work.go b/arkruntime/environment_work.go index 9c055ac..619d4a9 100644 --- a/arkruntime/environment_work.go +++ b/arkruntime/environment_work.go @@ -138,6 +138,10 @@ func (c *Client) StopWork( if body.WorkID == "" { return errors.New("missing required work_id") } + force, forceSet := body.Force.Get() + if _, reasonSet := body.Reason.Get(); reasonSet && (!forceSet || !force) { + return errors.New("reason requires force=true") + } u := c.fullURL(fmt.Sprintf("%s/%s/work/%s/stop", environmentsPrefix, environment.PathEscape(body.EnvironmentID), @@ -148,15 +152,24 @@ func (c *Client) StopWork( } type stopWorkRequestBody struct { - Force *bool `json:"force,omitempty"` + Force *bool `json:"force,omitempty"` + Reason *environment.WorkStopReason `json:"reason,omitempty"` } func stopWorkBody(body *environment.StopWorkRequest) any { + request := stopWorkRequestBody{} force, ok := body.Force.Get() - if !ok { + if ok { + request.Force = &force + } + reason, ok := body.Reason.Get() + if ok { + request.Reason = &reason + } + if request.Force == nil && request.Reason == nil { return nil } - return stopWorkRequestBody{Force: &force} + return request } func (c *Client) doControlPlaneRequest( diff --git a/arkruntime/lib/environments/integration_test.go b/arkruntime/lib/environments/integration_test.go index e4c012e..39f2017 100644 --- a/arkruntime/lib/environments/integration_test.go +++ b/arkruntime/lib/environments/integration_test.go @@ -5,6 +5,7 @@ package environments import ( "context" "encoding/json" + "errors" "io" "net/http" "strings" @@ -12,11 +13,37 @@ import ( "testing" "time" + "github.com/volcengine/ark-runtime-go/arkruntime/model/environment" selfhosted "github.com/volcengine/ark-runtime-go/arkruntime/selfhosted" "github.com/volcengine/ark-runtime-go/arkruntime/tools/agenttoolset" "github.com/volcengine/ark-runtime-go/arkruntime/toolset" ) +func TestStopReasonForExit(t *testing.T) { + tests := []struct { + name string + err error + cause heartbeatStopCause + want environment.WorkStopReason + wantValue bool + }{ + {name: "completed", want: environment.WorkStopReasonCompleted, wantValue: true}, + {name: "idle timeout", err: selfhosted.ErrIdleTimeout, want: environment.WorkStopReasonCompleted, wantValue: true}, + {name: "session terminated", err: selfhosted.ErrSessionTerminated, want: environment.WorkStopReasonCompleted, wantValue: true}, + {name: "external cancellation", err: context.Canceled, want: environment.WorkStopReasonWorkerAbnormal, wantValue: true}, + {name: "worker abnormal", err: errors.New("worker failed"), want: environment.WorkStopReasonWorkerAbnormal, wantValue: true}, + {name: "platform stop", err: context.Canceled, cause: heartbeatStopCauseStopRequested}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := stopReasonForExit(tt.err, tt.cause).Get() + if ok != tt.wantValue || got != tt.want { + t.Fatalf("stop reason = %q, set=%v", got, ok) + } + }) + } +} + type fakeEnvironmentWorkerAPI struct { mu sync.Mutex events []selfhosted.Event @@ -29,6 +56,7 @@ type fakeEnvironmentWorkerAPI struct { heartbeat func(context.Context, selfhosted.HeartbeatWorkRequest) (*selfhosted.HeartbeatResponse, error) getSession func(context.Context, selfhosted.GetSessionRequest) (*selfhosted.Session, error) + listEvents func(context.Context, selfhosted.ListEventsRequest) (*selfhosted.ListEventsResponse, error) onStop func() } @@ -86,7 +114,10 @@ func (f *fakeEnvironmentWorkerAPI) GetSession(ctx context.Context, req selfhoste return &selfhosted.Session{ID: req.SessionID}, nil } -func (f *fakeEnvironmentWorkerAPI) ListEvents(context.Context, selfhosted.ListEventsRequest) (*selfhosted.ListEventsResponse, error) { +func (f *fakeEnvironmentWorkerAPI) ListEvents(ctx context.Context, req selfhosted.ListEventsRequest) (*selfhosted.ListEventsResponse, error) { + if f.listEvents != nil { + return f.listEvents(ctx, req) + } f.mu.Lock() defer f.mu.Unlock() if len(f.events) == 0 { @@ -97,6 +128,57 @@ func (f *fakeEnvironmentWorkerAPI) ListEvents(context.Context, selfhosted.ListEv return &selfhosted.ListEventsResponse{Events: events}, nil } +func TestEnvironmentWorkerExternalCancellationReportsWorkerAbnormal(t *testing.T) { + listStarted := make(chan struct{}) + api := &fakeEnvironmentWorkerAPI{ + listEvents: func(ctx context.Context, _ selfhosted.ListEventsRequest) (*selfhosted.ListEventsResponse, error) { + close(listStarted) + <-ctx.Done() + return nil, ctx.Err() + }, + } + worker := NewEnvironmentWorker(api, EnvironmentWorkerOptions{ + EnvironmentID: "env_local", + WorkerID: "worker_local", + Workdir: t.TempDir(), + }) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + done := make(chan error, 1) + go func() { + done <- worker.HandleItem(ctx, HandleItemOptions{ + WorkID: "work_local", + EnvironmentID: "env_local", + SessionID: "sess_local", + }) + }() + + select { + case <-listStarted: + case <-time.After(time.Second): + t.Fatal("worker did not start event polling") + } + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("HandleItem err = %v", err) + } + case <-time.After(time.Second): + t.Fatal("worker did not stop after context cancellation") + } + + api.mu.Lock() + defer api.mu.Unlock() + if len(api.stops) != 1 { + t.Fatalf("stop count = %d, want 1", len(api.stops)) + } + reason, ok := api.stops[0].Reason.Get() + if !ok || reason != environment.WorkStopReasonWorkerAbnormal { + t.Fatalf("stop reason = %q, set=%v, want worker_abnormal", reason, ok) + } +} + func (f *fakeEnvironmentWorkerAPI) SendEvent(_ context.Context, req selfhosted.SendEventRequest) error { f.mu.Lock() defer f.mu.Unlock() @@ -156,6 +238,9 @@ func TestEnvironmentWorkerRunHandlesPolledWorkInProcess(t *testing.T) { if force, ok := api.stops[0].Force.Get(); !ok || !force { t.Fatalf("worker stop should be force=true: %+v", api.stops[0]) } + if reason, ok := api.stops[0].Reason.Get(); !ok || reason != environment.WorkStopReasonCompleted { + t.Fatalf("worker stop reason = %q, want completed", reason) + } if len(api.sent) != 1 { t.Fatalf("sent events = %+v", api.sent) } @@ -305,6 +390,8 @@ func TestEnvironmentWorkerStopsWorkOnSessionIdleEvent(t *testing.T) { t.Fatalf("worker stop = %+v", got) } else if force, ok := got.Force.Get(); !ok || !force { t.Fatalf("worker stop should be force=true: %+v", got) + } else if reason, ok := got.Reason.Get(); !ok || reason != environment.WorkStopReasonCompleted { + t.Fatalf("worker stop reason = %q, want completed", reason) } if len(api.sent) != 0 { t.Fatalf("sent events = %+v", api.sent) diff --git a/arkruntime/lib/environments/poller.go b/arkruntime/lib/environments/poller.go index d37ab4b..86cf46a 100644 --- a/arkruntime/lib/environments/poller.go +++ b/arkruntime/lib/environments/poller.go @@ -224,6 +224,7 @@ func (p *WorkPoller) discardInvalidWork(item selfhosted.WorkItem, _ string) { EnvironmentID: item.EnvironmentID, WorkID: item.ID, Force: environment.NewOptBool(true), + Reason: environment.NewOptWorkStopReason(environment.WorkStopReasonOthers), }); err != nil && !isResolvedStatus(err) { p.logger.Warn("stop invalid work failed", "work_id", item.ID, "err", err) } diff --git a/arkruntime/lib/environments/poller_test.go b/arkruntime/lib/environments/poller_test.go index 1f3546c..26ba1a5 100644 --- a/arkruntime/lib/environments/poller_test.go +++ b/arkruntime/lib/environments/poller_test.go @@ -207,6 +207,28 @@ func TestWorkPollerStopsPollingOnPermanentAckFailure(t *testing.T) { } } +func TestWorkPollerMarksInvalidWorkAsOther(t *testing.T) { + api := &fakePollerAPI{ + pollItem: newTestWorkItem(testWorkID, "env_1", ""), + } + poller := NewWorkPoller(context.Background(), api, WorkPollerOptions{ + EnvironmentID: "env_1", + WorkerID: "worker_1", + Drain: true, + }) + + if poller.Next() { + t.Fatal("Next should discard work without a session id") + } + if api.stopCount != 1 { + t.Fatalf("stop_count=%d", api.stopCount) + } + reason, ok := api.stops[0].Reason.Get() + if !ok || reason != environment.WorkStopReasonOthers { + t.Fatalf("stop reason = %q, set=%v", reason, ok) + } +} + func TestWorkPollerTreatsEmptyWorkIDAsEmptyPoll(t *testing.T) { api := &fakePollerAPI{ pollItem: newTestWorkItem("", "env_1", testSessionID), diff --git a/arkruntime/lib/environments/worker.go b/arkruntime/lib/environments/worker.go index da74ef7..1c968f9 100644 --- a/arkruntime/lib/environments/worker.go +++ b/arkruntime/lib/environments/worker.go @@ -172,7 +172,11 @@ func (w *EnvironmentWorker) handleItem(ctx context.Context, work claimedWork) (e <-heartbeatDone cause := loadHeartbeatStopCause(&heartbeatCause) if shouldStopItem(cause) { - _ = w.stopItem(api, work) + exitErr := err + if exitErr == nil { + exitErr = ctx.Err() + } + _ = w.stopItem(api, work, stopReasonForExit(exitErr, cause)) } else { logger.Info("skip stop work after heartbeat ownership became uncertain", "cause", cause) } @@ -301,13 +305,18 @@ func (w *EnvironmentWorker) workdir() (string, error) { return filepath.Abs(root) } -func (w *EnvironmentWorker) stopItem(api selfhosted.API, work claimedWork) error { +func (w *EnvironmentWorker) stopItem( + api selfhosted.API, + work claimedWork, + reason environment.OptWorkStopReason, +) error { stopCtx, stopCancel := context.WithTimeout(context.Background(), stopTimeout) defer stopCancel() req := selfhosted.StopWorkRequest{ EnvironmentID: work.EnvironmentID, WorkID: work.ID, Force: environment.NewOptBool(true), + Reason: reason, } if err := api.StopWork(stopCtx, req); err != nil { if selfhosted.IsStatus(err, 409) || selfhosted.IsStatus(err, 412) { @@ -320,6 +329,16 @@ func (w *EnvironmentWorker) stopItem(api selfhosted.API, work claimedWork) error return nil } +func stopReasonForExit(err error, cause heartbeatStopCause) environment.OptWorkStopReason { + if cause == heartbeatStopCauseStopRequested { + return environment.OptWorkStopReason{} + } + if err == nil || errors.Is(err, selfhosted.ErrIdleTimeout) || errors.Is(err, selfhosted.ErrSessionTerminated) { + return environment.NewOptWorkStopReason(environment.WorkStopReasonCompleted) + } + return environment.NewOptWorkStopReason(environment.WorkStopReasonWorkerAbnormal) +} + func (w *EnvironmentWorker) logger() *selfhostedlog.Logger { return selfhostedlog.New(w.opts.Logger) } diff --git a/arkruntime/model/environment/oas_json_gen.go b/arkruntime/model/environment/oas_json_gen.go index 93d35bc..6761fa8 100644 --- a/arkruntime/model/environment/oas_json_gen.go +++ b/arkruntime/model/environment/oas_json_gen.go @@ -384,6 +384,12 @@ 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") @@ -392,13 +398,14 @@ func (s *EnvConfig) encodeFields(e *jx.Encoder) { } } -var jsonFieldsNameOfEnvConfig = [6]string{ +var jsonFieldsNameOfEnvConfig = [7]string{ 0: "type", 1: "networking", 2: "packages", 3: "env", 4: "setup_script", - 5: "tos", + 5: "active_recovery", + 6: "tos", } // Decode decodes EnvConfig from json. @@ -460,6 +467,16 @@ 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() @@ -1739,6 +1756,41 @@ 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 { @@ -1973,6 +2025,39 @@ func (s *OptUpdateEnvironmentRequestMetadata) UnmarshalJSON(data []byte) error { return s.Decode(d) } +// Encode encodes WorkStopReason as json. +func (o OptWorkStopReason) Encode(e *jx.Encoder) { + if !o.Set { + return + } + e.Str(string(o.Value)) +} + +// Decode decodes WorkStopReason from json. +func (o *OptWorkStopReason) Decode(d *jx.Decoder) error { + if o == nil { + return errors.New("invalid: unable to decode OptWorkStopReason to nil") + } + o.Set = true + if err := o.Value.Decode(d); err != nil { + return err + } + return nil +} + +// MarshalJSON implements stdjson.Marshaler. +func (s OptWorkStopReason) MarshalJSON() ([]byte, error) { + e := jx.Encoder{} + s.Encode(&e) + return e.Bytes(), nil +} + +// UnmarshalJSON implements stdjson.Unmarshaler. +func (s *OptWorkStopReason) UnmarshalJSON(data []byte) error { + d := jx.DecodeBytes(data) + return s.Decode(d) +} + // Encode implements json.Marshaler. func (s *PackagesConfig) Encode(e *jx.Encoder) { e.ObjStart() @@ -2269,10 +2354,17 @@ func (s *StopWorkBody) encodeFields(e *jx.Encoder) { s.Force.Encode(e) } } + { + if s.Reason.Set { + e.FieldStart("reason") + s.Reason.Encode(e) + } + } } -var jsonFieldsNameOfStopWorkBody = [1]string{ +var jsonFieldsNameOfStopWorkBody = [2]string{ 0: "force", + 1: "reason", } // Decode decodes StopWorkBody from json. @@ -2293,6 +2385,16 @@ func (s *StopWorkBody) Decode(d *jx.Decoder) error { }(); err != nil { return errors.Wrap(err, "decode field \"force\"") } + case "reason": + if err := func() error { + s.Reason.Reset() + if err := s.Reason.Decode(d); err != nil { + return err + } + return nil + }(); err != nil { + return errors.Wrap(err, "decode field \"reason\"") + } default: return d.Skip() } @@ -2885,13 +2987,25 @@ func (s *WorkItem) encodeFields(e *jx.Encoder) { s.StoppedAt.Encode(e) } } + { + if s.StopReason.Set { + e.FieldStart("stop_reason") + 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 = [13]string{ +var jsonFieldsNameOfWorkItem = [15]string{ 0: "id", 1: "acknowledged_at", 2: "created_at", @@ -2904,7 +3018,9 @@ var jsonFieldsNameOfWorkItem = [13]string{ 9: "state", 10: "stop_requested_at", 11: "stopped_at", - 12: "type", + 12: "stop_reason", + 13: "recovery_count", + 14: "type", } // Decode decodes WorkItem from json. @@ -3049,8 +3165,28 @@ func (s *WorkItem) Decode(d *jx.Decoder) error { }(); err != nil { return errors.Wrap(err, "decode field \"stopped_at\"") } + case "stop_reason": + if err := func() error { + s.StopReason.Reset() + if err := s.StopReason.Decode(d); err != nil { + return err + } + return nil + }(); 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 << 4 + requiredBitSet[1] |= 1 << 6 if err := func() error { if err := s.Type.Decode(d); err != nil { return err @@ -3070,7 +3206,7 @@ func (s *WorkItem) Decode(d *jx.Decoder) error { var failures []validate.FieldError for i, mask := range [2]uint8{ 0b00011101, - 0b00010010, + 0b01000010, } { if result := (requiredBitSet[i] & mask) ^ mask; result != 0 { // Mask only required fields and check equality to mask using XOR. @@ -3199,3 +3335,51 @@ func (s *WorkState) UnmarshalJSON(data []byte) error { d := jx.DecodeBytes(data) return s.Decode(d) } + +// Encode encodes WorkStopReason as json. +func (s WorkStopReason) Encode(e *jx.Encoder) { + e.Str(string(s)) +} + +// Decode decodes WorkStopReason from json. +func (s *WorkStopReason) Decode(d *jx.Decoder) error { + if s == nil { + return errors.New("invalid: unable to decode WorkStopReason to nil") + } + v, err := d.StrBytes() + if err != nil { + return err + } + // Try to use constant string. + switch WorkStopReason(v) { + case WorkStopReasonLeaseExpired: + *s = WorkStopReasonLeaseExpired + case WorkStopReasonWorkerAbnormal: + *s = WorkStopReasonWorkerAbnormal + case WorkStopReasonUserCancelled: + *s = WorkStopReasonUserCancelled + case WorkStopReasonAdminStopped: + *s = WorkStopReasonAdminStopped + case WorkStopReasonCompleted: + *s = WorkStopReasonCompleted + case WorkStopReasonOthers: + *s = WorkStopReasonOthers + default: + *s = WorkStopReason(v) + } + + return nil +} + +// MarshalJSON implements stdjson.Marshaler. +func (s WorkStopReason) MarshalJSON() ([]byte, error) { + e := jx.Encoder{} + s.Encode(&e) + return e.Bytes(), nil +} + +// UnmarshalJSON implements stdjson.Unmarshaler. +func (s *WorkStopReason) UnmarshalJSON(data []byte) error { + d := jx.DecodeBytes(data) + return s.Decode(d) +} diff --git a/arkruntime/model/environment/oas_schemas_gen.go b/arkruntime/model/environment/oas_schemas_gen.go index 1e2fc6b..3048964 100644 --- a/arkruntime/model/environment/oas_schemas_gen.go +++ b/arkruntime/model/environment/oas_schemas_gen.go @@ -128,6 +128,15 @@ 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"` } @@ -157,6 +166,11 @@ 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 @@ -187,6 +201,11 @@ 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 @@ -1380,6 +1399,52 @@ func (o OptUpdateEnvironmentRequestMetadata) Or(d UpdateEnvironmentRequestMetada return d } +// NewOptWorkStopReason returns new OptWorkStopReason with value set to v. +func NewOptWorkStopReason(v WorkStopReason) OptWorkStopReason { + return OptWorkStopReason{ + Value: v, + Set: true, + } +} + +// OptWorkStopReason is optional WorkStopReason. +type OptWorkStopReason struct { + Value WorkStopReason + Set bool +} + +// IsSet returns true if OptWorkStopReason was set. +func (o OptWorkStopReason) IsSet() bool { return o.Set } + +// Reset unsets value. +func (o *OptWorkStopReason) Reset() { + var v WorkStopReason + o.Value = v + o.Set = false +} + +// SetTo sets value to v. +func (o *OptWorkStopReason) SetTo(v WorkStopReason) { + o.Set = true + o.Value = v +} + +// Get returns value and boolean that denotes whether value was set. +func (o OptWorkStopReason) Get() (v WorkStopReason, ok bool) { + if !o.Set { + return v, false + } + return o.Value, true +} + +// Or returns value if set, or given parameter if does not. +func (o OptWorkStopReason) Or(d WorkStopReason) WorkStopReason { + if v, ok := o.Get(); ok { + return v + } + return d +} + // 启动时预装的依赖包。版本通过包管理器自身语义指定 (e.g. pip 用 // `"pkg==1.0.0"`；不指定版本则装最新)。. // Ref: #/components/schemas/PackagesConfig @@ -1510,6 +1575,16 @@ func (s *PackagesConfigType) UnmarshalText(data []byte) error { type StopWorkBody struct { // 是否强制停止。. Force OptBool `json:"force"` + // 强制停止原因；仅允许与 `force=true` 一起提交，省略时按 `others` 记录。 + // Worker 正常完成时传 `completed`，异常退出时传 + // `worker_abnormal`；用户或管理员操作分别传 + // `user_cancelled` 或 `admin_stopped`，无法归类时传 `others`。`lease_expired` + // 仅由平台在租约 + // 过期时写入，不应通过 StopWork 上报。该字段用于停止诊断，并在 Worker + // 上报 `completed` + // 且全部 tool_use + // 已闭合时作为清零恢复次数的信号；平台不会按停止原因排除积极恢复。. + Reason OptWorkStopReason `json:"reason"` } // GetForce returns the value of Force. @@ -1517,11 +1592,21 @@ func (s *StopWorkBody) GetForce() OptBool { return s.Force } +// GetReason returns the value of Reason. +func (s *StopWorkBody) GetReason() OptWorkStopReason { + return s.Reason +} + // SetForce sets the value of Force. func (s *StopWorkBody) SetForce(val OptBool) { s.Force = val } +// SetReason sets the value of Reason. +func (s *StopWorkBody) SetReason(val OptWorkStopReason) { + s.Reason = val +} + // Environment 产物存储位置。设置后 outputs 文件会注册到用户指定的 TOS // bucket/prefix；不设置则走方舟默认存储。. // Ref: #/components/schemas/TosConfig @@ -1718,6 +1803,16 @@ type WorkItem struct { StopRequestedAt OptString `json:"stop_requested_at"` // Work 停止时间，RFC 3339。. StoppedAt OptString `json:"stopped_at"` + // 最近一次停止原因。首次停止前省略；state 为 stopping / stopped + // 时返回本次停止原因； + // 自动恢复到 queued / starting / active + // 后仍保留上一次停止原因，当前生命周期状态以 + // `state` 为准。. + StopReason OptWorkStopReason `json:"stop_reason"` + // 自上次成功完成且全部 tool_use 已闭合后，平台已自动将本 Work + // 重新排队的次数； + // 首次执行或字段省略时按 0 处理。手动 Ensure 不会清零该计数。. + RecoveryCount OptInt32 `json:"recovery_count"` // 对象类型，固定为 `work`。. Type WorkItemType `json:"type"` } @@ -1782,6 +1877,16 @@ func (s *WorkItem) GetStoppedAt() OptString { return s.StoppedAt } +// GetStopReason returns the value of StopReason. +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 @@ -1847,6 +1952,16 @@ func (s *WorkItem) SetStoppedAt(val OptString) { s.StoppedAt = val } +// SetStopReason sets the value of StopReason. +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 @@ -1950,3 +2065,74 @@ func (s *WorkState) UnmarshalText(data []byte) error { return errors.Errorf("invalid value: %q", data) } } + +// Work 停止原因。. +// Ref: #/components/schemas/WorkStopReason +type WorkStopReason string + +const ( + WorkStopReasonLeaseExpired WorkStopReason = "lease_expired" + WorkStopReasonWorkerAbnormal WorkStopReason = "worker_abnormal" + WorkStopReasonUserCancelled WorkStopReason = "user_cancelled" + WorkStopReasonAdminStopped WorkStopReason = "admin_stopped" + WorkStopReasonCompleted WorkStopReason = "completed" + WorkStopReasonOthers WorkStopReason = "others" +) + +// AllValues returns all WorkStopReason values. +func (WorkStopReason) AllValues() []WorkStopReason { + return []WorkStopReason{ + WorkStopReasonLeaseExpired, + WorkStopReasonWorkerAbnormal, + WorkStopReasonUserCancelled, + WorkStopReasonAdminStopped, + WorkStopReasonCompleted, + WorkStopReasonOthers, + } +} + +// MarshalText implements encoding.TextMarshaler. +func (s WorkStopReason) MarshalText() ([]byte, error) { + switch s { + case WorkStopReasonLeaseExpired: + return []byte(s), nil + case WorkStopReasonWorkerAbnormal: + return []byte(s), nil + case WorkStopReasonUserCancelled: + return []byte(s), nil + case WorkStopReasonAdminStopped: + return []byte(s), nil + case WorkStopReasonCompleted: + return []byte(s), nil + case WorkStopReasonOthers: + return []byte(s), nil + default: + return nil, errors.Errorf("invalid value: %q", s) + } +} + +// UnmarshalText implements encoding.TextUnmarshaler. +func (s *WorkStopReason) UnmarshalText(data []byte) error { + switch WorkStopReason(data) { + case WorkStopReasonLeaseExpired: + *s = WorkStopReasonLeaseExpired + return nil + case WorkStopReasonWorkerAbnormal: + *s = WorkStopReasonWorkerAbnormal + return nil + case WorkStopReasonUserCancelled: + *s = WorkStopReasonUserCancelled + return nil + case WorkStopReasonAdminStopped: + *s = WorkStopReasonAdminStopped + return nil + case WorkStopReasonCompleted: + *s = WorkStopReasonCompleted + return nil + case WorkStopReasonOthers: + *s = WorkStopReasonOthers + return nil + default: + return errors.Errorf("invalid value: %q", data) + } +} diff --git a/arkruntime/model/environment/oas_validators_gen.go b/arkruntime/model/environment/oas_validators_gen.go index 1fa0e44..f760c45 100644 --- a/arkruntime/model/environment/oas_validators_gen.go +++ b/arkruntime/model/environment/oas_validators_gen.go @@ -365,6 +365,36 @@ func (s PackagesConfigType) Validate() error { } } +func (s *StopWorkBody) Validate() error { + if s == nil { + return validate.ErrNilPointer + } + + var failures []validate.FieldError + if err := func() error { + if value, ok := s.Reason.Get(); ok { + if err := func() error { + if err := value.Validate(); err != nil { + return err + } + return nil + }(); err != nil { + return err + } + } + return nil + }(); err != nil { + failures = append(failures, validate.FieldError{ + Name: "reason", + Error: err, + }) + } + if len(failures) > 0 { + return &validate.Error{Fields: failures} + } + return nil +} + func (s *UpdateEnvironmentRequest) Validate() error { if s == nil { return validate.ErrNilPointer @@ -430,6 +460,24 @@ func (s *WorkItem) Validate() error { Error: err, }) } + if err := func() error { + if value, ok := s.StopReason.Get(); ok { + if err := func() error { + if err := value.Validate(); err != nil { + return err + } + return nil + }(); err != nil { + return err + } + } + return nil + }(); err != nil { + failures = append(failures, validate.FieldError{ + Name: "stop_reason", + Error: err, + }) + } if err := func() error { if err := s.Type.Validate(); err != nil { return err @@ -472,3 +520,22 @@ func (s WorkState) Validate() error { return errors.Errorf("invalid value: %v", s) } } + +func (s WorkStopReason) Validate() error { + switch s { + case "lease_expired": + return nil + case "worker_abnormal": + return nil + case "user_cancelled": + return nil + case "admin_stopped": + return nil + case "completed": + return nil + case "others": + return nil + default: + return errors.Errorf("invalid value: %v", s) + } +} diff --git a/arkruntime/model/environment/work_shim.go b/arkruntime/model/environment/work_shim.go index 77c2108..8e72fa9 100644 --- a/arkruntime/model/environment/work_shim.go +++ b/arkruntime/model/environment/work_shim.go @@ -54,6 +54,8 @@ type StopWorkRequest struct { EnvironmentID string `json:"environment_id"` WorkID string `json:"work_id"` Force OptBool `json:"force,omitempty"` + // Reason is only valid when Force is true. + Reason OptWorkStopReason `json:"reason,omitempty"` } // SessionIDValue returns the session id carried by the work item. diff --git a/arkruntime/self_hosted_client_test.go b/arkruntime/self_hosted_client_test.go index 0a1208a..5b06e13 100644 --- a/arkruntime/self_hosted_client_test.go +++ b/arkruntime/self_hosted_client_test.go @@ -45,7 +45,20 @@ func TestEnvironmentWorkRequests(t *testing.T) { if got := r.URL.Query().Get("max_items"); got != "" { t.Fatalf("unexpected max_items query = %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","latest_heartbeat_at":"2026-08-10T00:00:00Z","state":"queued","type":"work"}`)) + pollResponse := map[string]any{ + "id": "work-1", + "created_at": "2026-08-10T00:00:00Z", + "data": map[string]any{"id": "sess-1", "type": "session"}, + "environment_id": "env-1", + "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 { + t.Errorf("encode poll response: %v", err) + } case "POST /environments/env-1/work/work-1/ack": assertNoBody(t, r) if got := r.Header.Get(environmentWorkWorkerIDHeader); got != "worker-1" { @@ -62,9 +75,9 @@ func TestEnvironmentWorkRequests(t *testing.T) { } _, _ = w.Write([]byte(`{"last_heartbeat":"2026-08-10T00:00:00Z","lease_extended":true,"state":"active","ttl_seconds":30,"type":"work_heartbeat"}`)) case "POST /environments/env-1/work/work-1/stop": - var body map[string]bool + var body map[string]any decodeJSONBody(t, r, &body) - if len(body) != 1 || !body["force"] { + if len(body) != 2 || body["force"] != true || body["reason"] != "worker_abnormal" { t.Fatalf("stop body = %+v", body) } _, _ = w.Write([]byte(`{"id":"work-1","created_at":"2026-08-10T00:00:00Z","data":{"id":"sess-1","type":"session"},"environment_id":"env-1","state":"stopping","type":"work"}`)) @@ -94,6 +107,9 @@ 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", @@ -116,10 +132,18 @@ func TestEnvironmentWorkRequests(t *testing.T) { if heartbeat.TTLSeconds != 30 { t.Fatalf("HeartbeatWork() = %+v", heartbeat) } + if err := client.StopWork(ctx, &environment.StopWorkRequest{ + EnvironmentID: "env-1", + WorkID: "work-invalid", + Reason: environment.NewOptWorkStopReason(environment.WorkStopReasonWorkerAbnormal), + }); err == nil || err.Error() != "reason requires force=true" { + t.Fatalf("StopWork() error = %v", err) + } if err := client.StopWork(ctx, &environment.StopWorkRequest{ EnvironmentID: "env-1", WorkID: "work-1", Force: environment.NewOptBool(true), + Reason: environment.NewOptWorkStopReason(environment.WorkStopReasonWorkerAbnormal), }); err != nil { t.Fatalf("StopWork() error = %v", err) }