From 8ae67979ecbf7f2c716218ff9de5ef43eeec60a2 Mon Sep 17 00:00:00 2001 From: "ark-hand[bot]" Date: Thu, 8 Oct 2026 09:31:20 +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: dbc9b17b17b0f2649adb7113752d53cf09d155ff Hand-Written-Reason: No Ark-APIs provenance marker; treated as a hand-written source commit. Release-Version: 0.9.0 --- .../runtime/models/environment/EnvConfig.java | 39 +++++++- .../EnvironmentWorkPoll200Response.java | 76 ++++++++++++++- .../models/environment/StopWorkBody.java | 45 ++++++++- .../runtime/models/environment/WorkItem.java | 76 ++++++++++++++- .../models/environment/WorkStopReason.java | 64 +++++++++++++ .../runtime/selfhosted/EnvironmentWorker.java | 14 ++- .../runtime/selfhosted/SelfHostedClient.java | 11 +++ .../ark/runtime/selfhosted/WorkPoller.java | 9 +- .../selfhosted/EnvironmentWorkerTest.java | 92 ++++++++++++++++++- .../selfhosted/SelfHostedClientTest.java | 51 +++++++++- 10 files changed, 461 insertions(+), 16 deletions(-) create mode 100644 src/main/java/com/volcengine/ark/runtime/models/environment/WorkStopReason.java diff --git a/src/main/java/com/volcengine/ark/runtime/models/environment/EnvConfig.java b/src/main/java/com/volcengine/ark/runtime/models/environment/EnvConfig.java index 97c12f9..eb2560b 100644 --- a/src/main/java/com/volcengine/ark/runtime/models/environment/EnvConfig.java +++ b/src/main/java/com/volcengine/ark/runtime/models/environment/EnvConfig.java @@ -32,6 +32,7 @@ EnvConfig.JSON_PROPERTY_PACKAGES, EnvConfig.JSON_PROPERTY_ENV, EnvConfig.JSON_PROPERTY_SETUP_SCRIPT, + EnvConfig.JSON_PROPERTY_ACTIVE_RECOVERY, EnvConfig.JSON_PROPERTY_TOS }) @javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", comments = "Generator version: 7.21.0") @@ -56,6 +57,10 @@ public class EnvConfig { @javax.annotation.Nullable private String setupScript; + public static final String JSON_PROPERTY_ACTIVE_RECOVERY = "active_recovery"; + @javax.annotation.Nullable + private Boolean activeRecovery; + public static final String JSON_PROPERTY_TOS = "tos"; @javax.annotation.Nullable private TosConfig tos; @@ -196,6 +201,31 @@ public void setSetupScript(@javax.annotation.Nullable String setupScript) { this.setupScript = setupScript; } + public EnvConfig activeRecovery(@javax.annotation.Nullable Boolean activeRecovery) { + + this.activeRecovery = activeRecovery; + return this; + } + + /** + * 是否启用 Work 积极恢复；仅 self_hosted Environment 支持，省略时等同于 false。该配置 仅允许在 Environment 级设置，不允许通过 Session override/upgrade 覆盖。开启后，平台在 Work 已 stopped、仍有未闭合 agent.tool_use 且 recovery_count 少于 5 次时将 Work 重新排队， 不按 stop_reason 排除。只有 Worker 上报 completed 且全部 tool_use 已闭合后才会清零计数。 + * @return activeRecovery + */ + @javax.annotation.Nullable + @JsonProperty(value = JSON_PROPERTY_ACTIVE_RECOVERY, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + + public Boolean getActiveRecovery() { + return activeRecovery; + } + + + @JsonProperty(value = JSON_PROPERTY_ACTIVE_RECOVERY, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + public void setActiveRecovery(@javax.annotation.Nullable Boolean activeRecovery) { + this.activeRecovery = activeRecovery; + } + public EnvConfig tos(@javax.annotation.Nullable TosConfig tos) { this.tos = tos; @@ -236,12 +266,13 @@ public boolean equals(Object o) { Objects.equals(this.packages, envConfig.packages) && Objects.equals(this.env, envConfig.env) && Objects.equals(this.setupScript, envConfig.setupScript) && + Objects.equals(this.activeRecovery, envConfig.activeRecovery) && Objects.equals(this.tos, envConfig.tos); } @Override public int hashCode() { - return Objects.hash(type, networking, packages, env, setupScript, tos); + return Objects.hash(type, networking, packages, env, setupScript, activeRecovery, tos); } @Override @@ -253,6 +284,7 @@ public String toString() { sb.append(" packages: ").append(toIndentedString(packages)).append("\n"); sb.append(" env: ").append(toIndentedString(env)).append("\n"); sb.append(" setupScript: ").append(toIndentedString(setupScript)).append("\n"); + sb.append(" activeRecovery: ").append(toIndentedString(activeRecovery)).append("\n"); sb.append(" tos: ").append(toIndentedString(tos)).append("\n"); sb.append("}"); return sb.toString(); @@ -298,6 +330,10 @@ public EnvConfig.Builder setupScript(String setupScript) { this.instance.setupScript = setupScript; return this; } + public EnvConfig.Builder activeRecovery(Boolean activeRecovery) { + this.instance.activeRecovery = activeRecovery; + return this; + } public EnvConfig.Builder tos(TosConfig tos) { this.instance.tos = tos; return this; @@ -341,6 +377,7 @@ public EnvConfig.Builder toBuilder() { .packages(getPackages()) .env(getEnv()) .setupScript(getSetupScript()) + .activeRecovery(getActiveRecovery()) .tos(getTos()); } diff --git a/src/main/java/com/volcengine/ark/runtime/models/environment/EnvironmentWorkPoll200Response.java b/src/main/java/com/volcengine/ark/runtime/models/environment/EnvironmentWorkPoll200Response.java index 4c553ce..390c2c2 100644 --- a/src/main/java/com/volcengine/ark/runtime/models/environment/EnvironmentWorkPoll200Response.java +++ b/src/main/java/com/volcengine/ark/runtime/models/environment/EnvironmentWorkPoll200Response.java @@ -42,6 +42,8 @@ EnvironmentWorkPoll200Response.JSON_PROPERTY_STATE, EnvironmentWorkPoll200Response.JSON_PROPERTY_STOP_REQUESTED_AT, EnvironmentWorkPoll200Response.JSON_PROPERTY_STOPPED_AT, + EnvironmentWorkPoll200Response.JSON_PROPERTY_STOP_REASON, + EnvironmentWorkPoll200Response.JSON_PROPERTY_RECOVERY_COUNT, EnvironmentWorkPoll200Response.JSON_PROPERTY_TYPE }) @JsonTypeName("EnvironmentWork_poll_200_response") @@ -95,6 +97,14 @@ public class EnvironmentWorkPoll200Response { @javax.annotation.Nullable private String stoppedAt; + public static final String JSON_PROPERTY_STOP_REASON = "stop_reason"; + @javax.annotation.Nullable + private WorkStopReason stopReason; + + public static final String JSON_PROPERTY_RECOVERY_COUNT = "recovery_count"; + @javax.annotation.Nullable + private Integer recoveryCount; + /** * 对象类型，固定为 `work`。 */ @@ -443,6 +453,56 @@ public void setStoppedAt(@javax.annotation.Nullable String stoppedAt) { this.stoppedAt = stoppedAt; } + public EnvironmentWorkPoll200Response stopReason(@javax.annotation.Nullable WorkStopReason stopReason) { + + this.stopReason = stopReason; + return this; + } + + /** + * 最近一次停止原因。首次停止前省略；state 为 stopping / stopped 时返回本次停止原因； 自动恢复到 queued / starting / active 后仍保留上一次停止原因，当前生命周期状态以 `state` 为准。 + * @return stopReason + */ + @javax.annotation.Nullable + @JsonProperty(value = JSON_PROPERTY_STOP_REASON, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + + public WorkStopReason getStopReason() { + return stopReason; + } + + + @JsonProperty(value = JSON_PROPERTY_STOP_REASON, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + public void setStopReason(@javax.annotation.Nullable WorkStopReason stopReason) { + this.stopReason = stopReason; + } + + public EnvironmentWorkPoll200Response recoveryCount(@javax.annotation.Nullable Integer recoveryCount) { + + this.recoveryCount = recoveryCount; + return this; + } + + /** + * 自上次成功完成且全部 tool_use 已闭合后，平台已自动将本 Work 重新排队的次数； 首次执行或字段省略时按 0 处理。手动 Ensure 不会清零该计数。 + * @return recoveryCount + */ + @javax.annotation.Nullable + @JsonProperty(value = JSON_PROPERTY_RECOVERY_COUNT, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + + public Integer getRecoveryCount() { + return recoveryCount; + } + + + @JsonProperty(value = JSON_PROPERTY_RECOVERY_COUNT, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + public void setRecoveryCount(@javax.annotation.Nullable Integer recoveryCount) { + this.recoveryCount = recoveryCount; + } + public EnvironmentWorkPoll200Response type(@javax.annotation.Nonnull TypeEnum type) { this.type = type; @@ -490,12 +550,14 @@ public boolean equals(Object o) { Objects.equals(this.state, environmentWorkPoll200Response.state) && Objects.equals(this.stopRequestedAt, environmentWorkPoll200Response.stopRequestedAt) && Objects.equals(this.stoppedAt, environmentWorkPoll200Response.stoppedAt) && + Objects.equals(this.stopReason, environmentWorkPoll200Response.stopReason) && + Objects.equals(this.recoveryCount, environmentWorkPoll200Response.recoveryCount) && Objects.equals(this.type, environmentWorkPoll200Response.type); } @Override public int hashCode() { - return Objects.hash(id, acknowledgedAt, createdAt, data, environmentId, latestHeartbeatAt, tags, secret, startedAt, state, stopRequestedAt, stoppedAt, type); + return Objects.hash(id, acknowledgedAt, createdAt, data, environmentId, latestHeartbeatAt, tags, secret, startedAt, state, stopRequestedAt, stoppedAt, stopReason, recoveryCount, type); } @Override @@ -514,6 +576,8 @@ public String toString() { sb.append(" state: ").append(toIndentedString(state)).append("\n"); sb.append(" stopRequestedAt: ").append(toIndentedString(stopRequestedAt)).append("\n"); sb.append(" stoppedAt: ").append(toIndentedString(stoppedAt)).append("\n"); + sb.append(" stopReason: ").append(toIndentedString(stopReason)).append("\n"); + sb.append(" recoveryCount: ").append(toIndentedString(recoveryCount)).append("\n"); sb.append(" type: ").append(toIndentedString(type)).append("\n"); sb.append("}"); return sb.toString(); @@ -587,6 +651,14 @@ public EnvironmentWorkPoll200Response.Builder stoppedAt(String stoppedAt) { this.instance.stoppedAt = stoppedAt; return this; } + public EnvironmentWorkPoll200Response.Builder stopReason(WorkStopReason stopReason) { + this.instance.stopReason = stopReason; + return this; + } + public EnvironmentWorkPoll200Response.Builder recoveryCount(Integer recoveryCount) { + this.instance.recoveryCount = recoveryCount; + return this; + } public EnvironmentWorkPoll200Response.Builder type(TypeEnum type) { this.instance.type = type; return this; @@ -637,6 +709,8 @@ public EnvironmentWorkPoll200Response.Builder toBuilder() { .state(getState()) .stopRequestedAt(getStopRequestedAt()) .stoppedAt(getStoppedAt()) + .stopReason(getStopReason()) + .recoveryCount(getRecoveryCount()) .type(getType()); } diff --git a/src/main/java/com/volcengine/ark/runtime/models/environment/StopWorkBody.java b/src/main/java/com/volcengine/ark/runtime/models/environment/StopWorkBody.java index a36edd6..f2c1ebd 100644 --- a/src/main/java/com/volcengine/ark/runtime/models/environment/StopWorkBody.java +++ b/src/main/java/com/volcengine/ark/runtime/models/environment/StopWorkBody.java @@ -25,7 +25,8 @@ * Stop work 的请求体。 */ @JsonPropertyOrder({ - StopWorkBody.JSON_PROPERTY_FORCE + StopWorkBody.JSON_PROPERTY_FORCE, + StopWorkBody.JSON_PROPERTY_REASON }) @javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", comments = "Generator version: 7.21.0") public class StopWorkBody { @@ -33,6 +34,10 @@ public class StopWorkBody { @javax.annotation.Nullable private Boolean force; + public static final String JSON_PROPERTY_REASON = "reason"; + @javax.annotation.Nullable + private WorkStopReason reason; + public StopWorkBody() { } @@ -61,6 +66,31 @@ public void setForce(@javax.annotation.Nullable Boolean force) { this.force = force; } + public StopWorkBody reason(@javax.annotation.Nullable WorkStopReason reason) { + + this.reason = reason; + return this; + } + + /** + * 强制停止原因；仅允许与 `force=true` 一起提交，省略时按 `others` 记录。 Worker 正常完成时传 `completed`，异常退出时传 `worker_abnormal`；用户或管理员操作分别传 `user_cancelled` 或 `admin_stopped`，无法归类时传 `others`。`lease_expired` 仅由平台在租约 过期时写入，不应通过 StopWork 上报。该字段用于停止诊断，并在 Worker 上报 `completed` 且全部 tool_use 已闭合时作为清零恢复次数的信号；平台不会按停止原因排除积极恢复。 + * @return reason + */ + @javax.annotation.Nullable + @JsonProperty(value = JSON_PROPERTY_REASON, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + + public WorkStopReason getReason() { + return reason; + } + + + @JsonProperty(value = JSON_PROPERTY_REASON, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + public void setReason(@javax.annotation.Nullable WorkStopReason reason) { + this.reason = reason; + } + @Override public boolean equals(Object o) { @@ -71,12 +101,13 @@ public boolean equals(Object o) { return false; } StopWorkBody stopWorkBody = (StopWorkBody) o; - return Objects.equals(this.force, stopWorkBody.force); + return Objects.equals(this.force, stopWorkBody.force) && + Objects.equals(this.reason, stopWorkBody.reason); } @Override public int hashCode() { - return Objects.hash(force); + return Objects.hash(force, reason); } @Override @@ -84,6 +115,7 @@ public String toString() { StringBuilder sb = new StringBuilder(); sb.append("class StopWorkBody {\n"); sb.append(" force: ").append(toIndentedString(force)).append("\n"); + sb.append(" reason: ").append(toIndentedString(reason)).append("\n"); sb.append("}"); return sb.toString(); } @@ -112,6 +144,10 @@ public StopWorkBody.Builder force(Boolean force) { this.instance.force = force; return this; } + public StopWorkBody.Builder reason(WorkStopReason reason) { + this.instance.reason = reason; + return this; + } /** @@ -146,7 +182,8 @@ public static StopWorkBody.Builder builder() { */ public StopWorkBody.Builder toBuilder() { return new StopWorkBody.Builder() - .force(getForce()); + .force(getForce()) + .reason(getReason()); } diff --git a/src/main/java/com/volcengine/ark/runtime/models/environment/WorkItem.java b/src/main/java/com/volcengine/ark/runtime/models/environment/WorkItem.java index e8638b0..ec2da50 100644 --- a/src/main/java/com/volcengine/ark/runtime/models/environment/WorkItem.java +++ b/src/main/java/com/volcengine/ark/runtime/models/environment/WorkItem.java @@ -41,6 +41,8 @@ WorkItem.JSON_PROPERTY_STATE, WorkItem.JSON_PROPERTY_STOP_REQUESTED_AT, WorkItem.JSON_PROPERTY_STOPPED_AT, + WorkItem.JSON_PROPERTY_STOP_REASON, + WorkItem.JSON_PROPERTY_RECOVERY_COUNT, WorkItem.JSON_PROPERTY_TYPE }) @javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", comments = "Generator version: 7.21.0") @@ -93,6 +95,14 @@ public class WorkItem { @javax.annotation.Nullable private String stoppedAt; + public static final String JSON_PROPERTY_STOP_REASON = "stop_reason"; + @javax.annotation.Nullable + private WorkStopReason stopReason; + + public static final String JSON_PROPERTY_RECOVERY_COUNT = "recovery_count"; + @javax.annotation.Nullable + private Integer recoveryCount; + /** * 对象类型，固定为 `work`。 */ @@ -441,6 +451,56 @@ public void setStoppedAt(@javax.annotation.Nullable String stoppedAt) { this.stoppedAt = stoppedAt; } + public WorkItem stopReason(@javax.annotation.Nullable WorkStopReason stopReason) { + + this.stopReason = stopReason; + return this; + } + + /** + * 最近一次停止原因。首次停止前省略；state 为 stopping / stopped 时返回本次停止原因； 自动恢复到 queued / starting / active 后仍保留上一次停止原因，当前生命周期状态以 `state` 为准。 + * @return stopReason + */ + @javax.annotation.Nullable + @JsonProperty(value = JSON_PROPERTY_STOP_REASON, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + + public WorkStopReason getStopReason() { + return stopReason; + } + + + @JsonProperty(value = JSON_PROPERTY_STOP_REASON, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + public void setStopReason(@javax.annotation.Nullable WorkStopReason stopReason) { + this.stopReason = stopReason; + } + + public WorkItem recoveryCount(@javax.annotation.Nullable Integer recoveryCount) { + + this.recoveryCount = recoveryCount; + return this; + } + + /** + * 自上次成功完成且全部 tool_use 已闭合后，平台已自动将本 Work 重新排队的次数； 首次执行或字段省略时按 0 处理。手动 Ensure 不会清零该计数。 + * @return recoveryCount + */ + @javax.annotation.Nullable + @JsonProperty(value = JSON_PROPERTY_RECOVERY_COUNT, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + + public Integer getRecoveryCount() { + return recoveryCount; + } + + + @JsonProperty(value = JSON_PROPERTY_RECOVERY_COUNT, required = false) + @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS) + public void setRecoveryCount(@javax.annotation.Nullable Integer recoveryCount) { + this.recoveryCount = recoveryCount; + } + public WorkItem type(@javax.annotation.Nonnull TypeEnum type) { this.type = type; @@ -488,12 +548,14 @@ public boolean equals(Object o) { Objects.equals(this.state, workItem.state) && Objects.equals(this.stopRequestedAt, workItem.stopRequestedAt) && Objects.equals(this.stoppedAt, workItem.stoppedAt) && + Objects.equals(this.stopReason, workItem.stopReason) && + Objects.equals(this.recoveryCount, workItem.recoveryCount) && Objects.equals(this.type, workItem.type); } @Override public int hashCode() { - return Objects.hash(id, acknowledgedAt, createdAt, data, environmentId, latestHeartbeatAt, tags, secret, startedAt, state, stopRequestedAt, stoppedAt, type); + return Objects.hash(id, acknowledgedAt, createdAt, data, environmentId, latestHeartbeatAt, tags, secret, startedAt, state, stopRequestedAt, stoppedAt, stopReason, recoveryCount, type); } @Override @@ -512,6 +574,8 @@ public String toString() { sb.append(" state: ").append(toIndentedString(state)).append("\n"); sb.append(" stopRequestedAt: ").append(toIndentedString(stopRequestedAt)).append("\n"); sb.append(" stoppedAt: ").append(toIndentedString(stoppedAt)).append("\n"); + sb.append(" stopReason: ").append(toIndentedString(stopReason)).append("\n"); + sb.append(" recoveryCount: ").append(toIndentedString(recoveryCount)).append("\n"); sb.append(" type: ").append(toIndentedString(type)).append("\n"); sb.append("}"); return sb.toString(); @@ -585,6 +649,14 @@ public WorkItem.Builder stoppedAt(String stoppedAt) { this.instance.stoppedAt = stoppedAt; return this; } + public WorkItem.Builder stopReason(WorkStopReason stopReason) { + this.instance.stopReason = stopReason; + return this; + } + public WorkItem.Builder recoveryCount(Integer recoveryCount) { + this.instance.recoveryCount = recoveryCount; + return this; + } public WorkItem.Builder type(TypeEnum type) { this.instance.type = type; return this; @@ -635,6 +707,8 @@ public WorkItem.Builder toBuilder() { .state(getState()) .stopRequestedAt(getStopRequestedAt()) .stoppedAt(getStoppedAt()) + .stopReason(getStopReason()) + .recoveryCount(getRecoveryCount()) .type(getType()); } diff --git a/src/main/java/com/volcengine/ark/runtime/models/environment/WorkStopReason.java b/src/main/java/com/volcengine/ark/runtime/models/environment/WorkStopReason.java new file mode 100644 index 0000000..e8e3717 --- /dev/null +++ b/src/main/java/com/volcengine/ark/runtime/models/environment/WorkStopReason.java @@ -0,0 +1,64 @@ +// Copyright (c) 2026 ByteDance Ltd. and/or its affiliates. +// SPDX-License-Identifier: Apache-2.0 + +/* + * Ark Managed Agents Environment API + * No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator) + * + * The version of the OpenAPI document: 0.1.0 + * + * + * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). + * https://openapi-generator.tech + * Do not edit the class manually. + */ + + +package com.volcengine.ark.runtime.models.environment; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; + +/** + * Work 停止原因。 + */ +public enum WorkStopReason { + + LEASE_EXPIRED("lease_expired"), + + WORKER_ABNORMAL("worker_abnormal"), + + USER_CANCELLED("user_cancelled"), + + ADMIN_STOPPED("admin_stopped"), + + COMPLETED("completed"), + + OTHERS("others"); + + private String value; + + WorkStopReason(String value) { + this.value = value; + } + + @JsonValue + public String getValue() { + return value; + } + + @Override + public String toString() { + return String.valueOf(value); + } + + @JsonCreator + public static WorkStopReason fromValue(String value) { + for (WorkStopReason b : WorkStopReason.values()) { + if (b.value.equalsIgnoreCase(value)) { + return b; + } + } + throw new IllegalArgumentException("Unexpected value '" + value + "'"); + } +} diff --git a/src/main/java/com/volcengine/ark/runtime/selfhosted/EnvironmentWorker.java b/src/main/java/com/volcengine/ark/runtime/selfhosted/EnvironmentWorker.java index a798f5e..1f928a9 100644 --- a/src/main/java/com/volcengine/ark/runtime/selfhosted/EnvironmentWorker.java +++ b/src/main/java/com/volcengine/ark/runtime/selfhosted/EnvironmentWorker.java @@ -6,6 +6,7 @@ import com.volcengine.ark.runtime.models.environment.HeartbeatWorkResponse; import com.volcengine.ark.runtime.models.environment.WorkItem; import com.volcengine.ark.runtime.models.environment.WorkState; +import com.volcengine.ark.runtime.models.environment.WorkStopReason; import java.io.IOException; import java.net.InetAddress; import java.nio.file.Files; @@ -90,6 +91,7 @@ private void handleItem(ClaimedWork work) throws IOException { AtomicReference heartbeatCause = new AtomicReference<>(""); Thread heartbeat = null; Initializer initializer = null; + WorkStopReason stopReason = WorkStopReason.COMPLETED; try { String workdir = workdir(); Thread heartbeatThread = new Thread( @@ -129,6 +131,12 @@ private void handleItem(ClaimedWork work) throws IOException { runner.close(); activeRunner = null; } + } catch (SessionToolRunner.IdleTimeoutException | SessionToolRunner.SessionTerminatedException e) { + // 预期的 session 生命周期结束保持 completed，避免落入异常退出分支。 + throw e; + } catch (IOException | RuntimeException | Error e) { + stopReason = WorkStopReason.WORKER_ABNORMAL; + throw e; } finally { if (initializer != null) { try { @@ -147,8 +155,12 @@ private void handleItem(ClaimedWork work) throws IOException { } String cause = heartbeatCause.get(); if (shouldStopItem(cause)) { + WorkStopReason reason = "stop_requested".equals(cause) ? null : stopReason; + if (closed.get() && cause.isEmpty()) { + reason = WorkStopReason.WORKER_ABNORMAL; + } try { - api.stopWork(work.environmentId, work.id, true); + api.stopWork(work.environmentId, work.id, true, reason); } catch (RuntimeException e) { if (!isResolvedStatus(e)) { options.logger.log(Level.WARNING, "stop work failed", e); diff --git a/src/main/java/com/volcengine/ark/runtime/selfhosted/SelfHostedClient.java b/src/main/java/com/volcengine/ark/runtime/selfhosted/SelfHostedClient.java index d1cdd97..dd04b25 100644 --- a/src/main/java/com/volcengine/ark/runtime/selfhosted/SelfHostedClient.java +++ b/src/main/java/com/volcengine/ark/runtime/selfhosted/SelfHostedClient.java @@ -11,6 +11,7 @@ import com.volcengine.ark.runtime.models.environment.HeartbeatWorkResponse; import com.volcengine.ark.runtime.models.environment.StopWorkBody; import com.volcengine.ark.runtime.models.environment.WorkItem; +import com.volcengine.ark.runtime.models.environment.WorkStopReason; import com.volcengine.ark.runtime.models.session.ManagedAgentsEventParams; import com.volcengine.ark.runtime.models.session.SendSessionEventsRequest; import com.volcengine.ark.runtime.models.skill.Skill; @@ -132,12 +133,22 @@ public HeartbeatWorkResponse heartbeatWork( } public void stopWork(String environmentId, String workId, boolean force) { + stopWork(environmentId, workId, force, null); + } + + public void stopWork(String environmentId, String workId, boolean force, WorkStopReason reason) { require(environmentId, "environmentId"); require(workId, "workId"); + if (reason != null && !force) { + throw new IllegalArgumentException("reason requires force=true"); + } StopWorkBody body = new StopWorkBody(); if (force) { body.setForce(Boolean.TRUE); } + if (reason != null) { + body.setReason(reason); + } execute(lifecycleApi.stopEnvironmentWork( environmentId, workId, body, Collections.emptyMap())); } diff --git a/src/main/java/com/volcengine/ark/runtime/selfhosted/WorkPoller.java b/src/main/java/com/volcengine/ark/runtime/selfhosted/WorkPoller.java index 18526ac..740f7e3 100644 --- a/src/main/java/com/volcengine/ark/runtime/selfhosted/WorkPoller.java +++ b/src/main/java/com/volcengine/ark/runtime/selfhosted/WorkPoller.java @@ -4,6 +4,7 @@ package com.volcengine.ark.runtime.selfhosted; import com.volcengine.ark.runtime.models.environment.WorkItem; +import com.volcengine.ark.runtime.models.environment.WorkStopReason; import java.util.Random; import java.util.logging.Level; import java.util.logging.Logger; @@ -95,7 +96,7 @@ public WorkItem next() { } current = item; if (options.autoStop) { - pendingStop = () -> stopItem(item, false); + pendingStop = () -> stopItem(item, false, null); } discards = 0; options.logger.info("claimed work work_id=" + item.getId() + " session_id=" + WorkItems.sessionId(item)); @@ -134,13 +135,13 @@ private void discardInvalidWork(WorkItem item) { options.logger.log(Level.WARNING, "ack invalid work failed", e); return; } - stopItem(item, true); + stopItem(item, true, WorkStopReason.OTHERS); backoffDiscard(); } - private void stopItem(WorkItem item, boolean force) { + private void stopItem(WorkItem item, boolean force, WorkStopReason reason) { try { - api.stopWork(item.getEnvironmentId(), item.getId(), force); + api.stopWork(item.getEnvironmentId(), item.getId(), force, reason); } catch (RuntimeException e) { if (!isResolvedStatus(e)) { options.logger.log(Level.WARNING, "stop work failed", e); diff --git a/src/test/java/com/volcengine/ark/runtime/selfhosted/EnvironmentWorkerTest.java b/src/test/java/com/volcengine/ark/runtime/selfhosted/EnvironmentWorkerTest.java index a7de386..cb89766 100644 --- a/src/test/java/com/volcengine/ark/runtime/selfhosted/EnvironmentWorkerTest.java +++ b/src/test/java/com/volcengine/ark/runtime/selfhosted/EnvironmentWorkerTest.java @@ -4,25 +4,32 @@ package com.volcengine.ark.runtime.selfhosted; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import com.volcengine.ark.runtime.models.environment.HeartbeatWorkResponse; import com.volcengine.ark.runtime.models.environment.WorkState; +import com.volcengine.ark.runtime.models.environment.WorkStopReason; import java.io.IOException; import java.lang.reflect.Method; import java.nio.file.Files; import java.nio.file.Path; +import java.util.Collections; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import okhttp3.MediaType; import okhttp3.OkHttpClient; import okhttp3.Protocol; import okhttp3.Request; import okhttp3.Response; import okhttp3.ResponseBody; +import okio.Buffer; import org.junit.Test; public class EnvironmentWorkerTest { @@ -64,6 +71,7 @@ public void emptyHeartbeatResponseDoesNotSpin() throws Exception { } assertEquals(1, client.heartbeats.get()); + assertEquals(WorkStopReason.WORKER_ABNORMAL, client.stopReason.get()); } @Test @@ -71,6 +79,7 @@ public void heartbeatStopCancelsRunnerBeforeEventPolling() throws Exception { CountDownLatch heartbeat = new CountDownLatch(1); AtomicInteger lists = new AtomicInteger(); AtomicInteger stops = new AtomicInteger(); + AtomicReference stopBody = new AtomicReference<>(); OkHttpClient http = new OkHttpClient.Builder().addInterceptor(chain -> { Request request = chain.request(); String path = request.url().encodedPath(); @@ -93,6 +102,9 @@ public void heartbeatStopCancelsRunnerBeforeEventPolling() throws Exception { } if (path.endsWith("/stop")) { stops.incrementAndGet(); + Buffer buffer = new Buffer(); + request.body().writeTo(buffer); + stopBody.set(buffer.readUtf8()); } return response(request, "{}"); }).build(); @@ -114,6 +126,35 @@ public void heartbeatStopCancelsRunnerBeforeEventPolling() throws Exception { assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started) < 1000L); assertEquals(0, lists.get()); assertEquals(1, stops.get()); + assertEquals("{\"force\":true}", stopBody.get()); + } + + @Test + public void externalCloseReportsWorkerAbnormal() throws Exception { + CloseClient client = new CloseClient(); + EnvironmentWorker worker = new EnvironmentWorker( + client, + new EnvironmentWorker.Options().workdir(Files.createTempDirectory("ark-java-worker-").toString())); + AtomicReference failure = new AtomicReference<>(); + Thread handling = new Thread(() -> { + try { + worker.handleItem(new EnvironmentWorker.HandleItemOptions() + .environmentId("env-1") + .workId("work-1") + .sessionId("session-1")); + } catch (Throwable error) { + failure.set(error); + } + }); + + handling.start(); + assertTrue(client.eventPolling.await(2, TimeUnit.SECONDS)); + worker.close(); + handling.join(2000L); + + assertFalse(handling.isAlive()); + assertNull(failure.get()); + assertEquals(WorkStopReason.WORKER_ABNORMAL, client.stopReason.get()); } @Test @@ -213,6 +254,7 @@ private static Response response(Request request, String body) throws IOExceptio private static class NullHeartbeatClient extends SelfHostedClient { private final CountDownLatch firstHeartbeat = new CountDownLatch(1); private final AtomicInteger heartbeats = new AtomicInteger(); + private final AtomicReference stopReason = new AtomicReference<>(); NullHeartbeatClient() { super("test-key"); @@ -239,7 +281,51 @@ public SessionSnapshot getSession(String sessionId) { } @Override - public void stopWork(String environmentId, String workId, boolean force) { + public void stopWork(String environmentId, String workId, boolean force, WorkStopReason reason) { + stopReason.set(reason); + } + } + + private static class CloseClient extends SelfHostedClient { + private final CountDownLatch eventPolling = new CountDownLatch(1); + private final AtomicReference stopReason = new AtomicReference<>(); + + CloseClient() { + super("test-key"); + } + + @Override + public HeartbeatWorkResponse heartbeatWork( + String environmentId, String workId, String expectedLastHeartbeat, int desiredTTLSeconds) { + return new HeartbeatWorkResponse() + .state(WorkState.ACTIVE) + .leaseExtended(Boolean.TRUE) + .lastHeartbeat("2026-09-20T00:00:00Z") + .ttlSeconds(30L); + } + + @Override + public SessionSnapshot getSession(String sessionId) { + SessionSnapshot session = new SessionSnapshot(); + session.setId(sessionId); + return session; + } + + @Override + public EventStream openEventStream(String sessionId) { + throw new SessionToolRunner.StreamUnsupportedException(); + } + + @Override + public ListEventsResponse listEvents( + String sessionId, String createdAtGt, String page, int limit, String order, List types) { + eventPolling.countDown(); + return new ListEventsResponse(Collections.emptyList(), ""); + } + + @Override + public void stopWork(String environmentId, String workId, boolean force, WorkStopReason reason) { + stopReason.set(reason); } } @@ -272,7 +358,7 @@ public SessionSnapshot getSession(String sessionId) { } @Override - public void stopWork(String environmentId, String workId, boolean force) { + public void stopWork(String environmentId, String workId, boolean force, WorkStopReason reason) { stops.incrementAndGet(); } } @@ -308,7 +394,7 @@ public SessionSnapshot getSession(String sessionId) { } @Override - public void stopWork(String environmentId, String workId, boolean force) { + public void stopWork(String environmentId, String workId, boolean force, WorkStopReason reason) { stops.incrementAndGet(); } } diff --git a/src/test/java/com/volcengine/ark/runtime/selfhosted/SelfHostedClientTest.java b/src/test/java/com/volcengine/ark/runtime/selfhosted/SelfHostedClientTest.java index 6b4cda5..3e9f1e4 100644 --- a/src/test/java/com/volcengine/ark/runtime/selfhosted/SelfHostedClientTest.java +++ b/src/test/java/com/volcengine/ark/runtime/selfhosted/SelfHostedClientTest.java @@ -12,6 +12,7 @@ import com.volcengine.ark.runtime.interceptor.RetryInterceptor; import com.volcengine.ark.runtime.models.environment.WorkItem; import com.volcengine.ark.runtime.models.environment.WorkState; +import com.volcengine.ark.runtime.models.environment.WorkStopReason; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; @@ -47,7 +48,9 @@ public void preservesNestedSessionWorkData() { String body = "{" + "\"id\":\"work-1\"," + "\"environment_id\":\"env-1\"," - + "\"data\":{\"id\":\"session-1\",\"type\":\"session\"}" + + "\"data\":{\"id\":\"session-1\",\"type\":\"session\"}," + + "\"stop_reason\":\"worker_abnormal\"," + + "\"recovery_count\":5" + "}"; OkHttpClient httpClient = new OkHttpClient.Builder() .addInterceptor(chain -> response(chain.request(), new AtomicReference<>(), body)) @@ -63,6 +66,8 @@ public void preservesNestedSessionWorkData() { assertEquals("work-1", item.getId()); assertEquals("env-1", item.getEnvironmentId()); assertEquals("session-1", item.getData().getId()); + assertEquals(WorkStopReason.WORKER_ABNORMAL, item.getStopReason()); + assertEquals(Integer.valueOf(5), item.getRecoveryCount()); assertEquals("session-1", WorkItems.sessionId(item)); } @@ -180,6 +185,50 @@ public void gracefulStopSendsEmptyJSONBody() throws Exception { assertEquals("{}", body.get()); } + @Test + public void stopWorkSerializesReason() throws Exception { + AtomicReference body = new AtomicReference<>(); + OkHttpClient httpClient = new OkHttpClient.Builder() + .addInterceptor(chain -> { + Buffer buffer = new Buffer(); + chain.request().body().writeTo(buffer); + body.set(buffer.readUtf8()); + return response(chain.request(), new AtomicReference<>()); + }) + .build(); + SelfHostedClient client = new SelfHostedClient.Builder() + .apiKey("test-api-key") + .httpClient(httpClient) + .build(); + + client.stopWork("env-1", "work-1", true, WorkStopReason.WORKER_ABNORMAL); + + assertEquals("{\"force\":true,\"reason\":\"worker_abnormal\"}", body.get()); + } + + @Test + public void stopWorkRejectsReasonWithoutForceBeforeSendingRequest() { + AtomicInteger calls = new AtomicInteger(); + OkHttpClient httpClient = new OkHttpClient.Builder() + .addInterceptor(chain -> { + calls.incrementAndGet(); + return response(chain.request(), new AtomicReference<>()); + }) + .build(); + SelfHostedClient client = new SelfHostedClient.Builder() + .apiKey("test-api-key") + .httpClient(httpClient) + .build(); + + try { + client.stopWork("env-1", "work-1", false, WorkStopReason.WORKER_ABNORMAL); + throw new AssertionError("expected invalid stop reason to be rejected"); + } catch (IllegalArgumentException error) { + assertEquals("reason requires force=true", error.getMessage()); + } + assertEquals(0, calls.get()); + } + @Test public void heartbeatUsesOneLeaseBoundedAttempt() { AtomicInteger calls = new AtomicInteger();