diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/TaskCancellation.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/TaskCancellation.java new file mode 100644 index 0000000000..7ed17b6136 --- /dev/null +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/TaskCancellation.java @@ -0,0 +1,154 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.harness.agent.subagent.task; + +import java.time.Duration; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +/** + * Typed acknowledgement for a background-task cancellation request. + * + *

{@link #requestStatus()} describes whether cancellation intent was accepted. It deliberately + * does not imply that the child execution has stopped. Callers that must wait before cleaning up + * parent resources can observe {@link #terminationSignal()} or call {@link + * #awaitTermination(Duration)} with their own time bound. + */ +public final class TaskCancellation { + + /** Outcome of recording and dispatching a cancellation request. */ + public enum RequestStatus { + ACCEPTED, + ALREADY_REQUESTED, + ALREADY_TERMINAL, + NOT_FOUND, + REJECTED + } + + /** Execution mechanism owned (or observed) by the task repository. */ + public enum ExecutionKind { + LOCAL, + REMOTE, + ADOPTED, + UNKNOWN + } + + /** Authoritative outcome of attempting to stop the underlying execution. */ + public enum TerminationStatus { + COOPERATIVE_STOP_CONFIRMED, + FORCED_STOP_CONFIRMED, + ALREADY_STOPPED, + DETACHED, + TIMED_OUT, + STOP_UNAVAILABLE, + CANCELLATION_FAILED + } + + /** Stable identity of the parent task and, where applicable, the remote child execution. */ + public record ExecutionIdentity( + String parentSessionId, + String taskId, + String subAgentId, + String childSessionId, + ExecutionKind executionKind, + String remoteTaskId) { + + public ExecutionIdentity { + Objects.requireNonNull(taskId, "taskId"); + Objects.requireNonNull(executionKind, "executionKind"); + } + } + + /** Completion value emitted only when execution stop is known, unavailable, or has failed. */ + public record Termination( + ExecutionIdentity identity, TerminationStatus status, String message) { + + public Termination { + Objects.requireNonNull(identity, "identity"); + Objects.requireNonNull(status, "status"); + } + + public boolean isStopConfirmed() { + return status == TerminationStatus.COOPERATIVE_STOP_CONFIRMED + || status == TerminationStatus.FORCED_STOP_CONFIRMED + || status == TerminationStatus.ALREADY_STOPPED; + } + } + + private final ExecutionIdentity identity; + private final RequestStatus requestStatus; + private final CompletableFuture termination; + + public TaskCancellation( + ExecutionIdentity identity, + RequestStatus requestStatus, + CompletableFuture termination) { + this.identity = Objects.requireNonNull(identity, "identity"); + this.requestStatus = Objects.requireNonNull(requestStatus, "requestStatus"); + this.termination = Objects.requireNonNull(termination, "termination"); + } + + public ExecutionIdentity identity() { + return identity; + } + + public RequestStatus requestStatus() { + return requestStatus; + } + + /** Whether the repository found the task and accepted this or an earlier cancel request. */ + public boolean isCancellationAccepted() { + return requestStatus == RequestStatus.ACCEPTED + || requestStatus == RequestStatus.ALREADY_REQUESTED; + } + + public CompletionStage terminationSignal() { + return termination; + } + + /** + * Wait for authoritative execution termination without imposing a repository-wide timeout. + * + *

A timeout is returned as a value rather than completing the shared signal: another caller + * may choose a longer bound and still receive the eventual authoritative result. + */ + public Termination awaitTermination(Duration timeout) throws InterruptedException { + Objects.requireNonNull(timeout, "timeout"); + if (timeout.isNegative()) { + throw new IllegalArgumentException("timeout must not be negative"); + } + try { + return termination.get(timeout.toMillis(), TimeUnit.MILLISECONDS); + } catch (TimeoutException e) { + return new Termination( + identity, + TerminationStatus.TIMED_OUT, + "Execution did not stop within " + timeout); + } catch (ExecutionException e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + return new Termination( + identity, + TerminationStatus.CANCELLATION_FAILED, + cause.getMessage() != null + ? cause.getMessage() + : cause.getClass().getSimpleName()); + } + } +} diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/TaskRepository.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/TaskRepository.java index 5376bf8772..5095a24969 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/TaskRepository.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/TaskRepository.java @@ -18,6 +18,7 @@ import io.agentscope.core.agent.RuntimeContext; import java.util.Collection; import java.util.List; +import java.util.concurrent.CompletableFuture; /** * Repository for managing background subagent tasks, scoped by session. @@ -90,6 +91,42 @@ BackgroundTask putTask( */ boolean cancelTask(RuntimeContext rc, String sessionId, String taskId); + /** + * Request cancellation and return a typed acknowledgement of underlying execution stop. + * + *

The default implementation preserves compatibility with repositories that only expose + * cancellation intent. Such repositories report {@link + * TaskCancellation.TerminationStatus#STOP_UNAVAILABLE}; implementations that own execution + * handles should override this method with an authoritative completion signal. + */ + default TaskCancellation cancelTaskWithAcknowledgement( + RuntimeContext rc, String sessionId, String taskId) { + boolean found = cancelTask(rc, sessionId, taskId); + TaskCancellation.ExecutionIdentity identity = + new TaskCancellation.ExecutionIdentity( + sessionId, + taskId, + null, + null, + TaskCancellation.ExecutionKind.UNKNOWN, + null); + CompletableFuture termination = + CompletableFuture.completedFuture( + new TaskCancellation.Termination( + identity, + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + found + ? "Repository accepted cancellation but cannot acknowledge" + + " execution stop" + : "Task was not found")); + return new TaskCancellation( + identity, + found + ? TaskCancellation.RequestStatus.ACCEPTED + : TaskCancellation.RequestStatus.NOT_FOUND, + termination); + } + // ------------------------------------------------------------------------ // Phase B-3 — push delivery API. // diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/WorkspaceTaskRepository.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/WorkspaceTaskRepository.java index e0ac9a22ed..b33145c053 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/WorkspaceTaskRepository.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/subagent/task/WorkspaceTaskRepository.java @@ -33,6 +33,8 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -109,6 +111,13 @@ public class WorkspaceTaskRepository implements TaskRepository { */ private final Map localTaskContexts = new ConcurrentHashMap<>(); + /** + * Execution lifecycle controls kept separately from result futures. Cancelling a + * {@link CompletableFuture} does not prove that its supplier has exited, so these controls + * expose a signal completed by the execution thread's {@code finally} block instead. + */ + private final Map executionControls = new ConcurrentHashMap<>(); + private final ExecutorService executor; private final boolean ownsExecutor; private final ScheduledExecutorService maintenanceScheduler; @@ -234,6 +243,19 @@ public BackgroundTask putTask( persistRecord(capturedRc, sessionId, record); String localKey = localKey(sessionId, taskId); + TaskCancellation.ExecutionKind executionKind = executionKind(spec); + ExecutionControl executionControl = + new ExecutionControl( + new TaskCancellation.ExecutionIdentity( + sessionId, + taskId, + subAgentId, + record.getSubSessionId(), + executionKind, + executionKind == TaskCancellation.ExecutionKind.REMOTE + ? taskId + : null)); + executionControls.put(localKey, executionControl); CompletableFuture future; if (spec instanceof TaskRunSpec.AdoptedTaskRunSpec adopted) { @@ -244,6 +266,9 @@ public BackgroundTask putTask( final String sid = sessionId; future.whenComplete( (result, err) -> { + if (!executionControl.isCancellationRequested()) { + executionControl.executionFinished(); + } if (err != null) { Throwable cause = err instanceof java.util.concurrent.CompletionException @@ -272,7 +297,8 @@ public BackgroundTask putTask( sessionId, taskId, subAgentId, - local.execution()), + local.execution(), + executionControl), executor); } else if (spec instanceof TaskRunSpec.RemoteTaskRunSpec remote) { future = @@ -284,7 +310,8 @@ public BackgroundTask putTask( taskId, subAgentId, remote, - true), + true, + executionControl), executor); } else { throw new IllegalArgumentException("Unsupported TaskRunSpec: " + spec.getClass()); @@ -302,16 +329,18 @@ private String runLocalSupplier( String sessionId, String taskId, String subAgentId, - Supplier taskExecution) { - Optional latest = - workspaceManager.readTaskRecord(rc, parentAgentId, sessionId, taskId); - if (latest.isPresent() && latest.get().isCancelRequested()) { - markCancelled(rc, sessionId, taskId); - return null; - } - - updateStatus(rc, sessionId, taskId, TaskStatus.RUNNING, null, null); + Supplier taskExecution, + ExecutionControl executionControl) { + executionControl.executionStarted(); try { + Optional latest = + workspaceManager.readTaskRecord(rc, parentAgentId, sessionId, taskId); + if (latest.isPresent() && latest.get().isCancelRequested()) { + markCancelled(rc, sessionId, taskId); + return null; + } + + updateStatus(rc, sessionId, taskId, TaskStatus.RUNNING, null, null); String result = taskExecution.get(); Optional afterRun = workspaceManager.readTaskRecord(rc, parentAgentId, sessionId, taskId); @@ -327,6 +356,8 @@ private String runLocalSupplier( updateStatus(rc, sessionId, taskId, TaskStatus.FAILED, null, errMsg); fireCompletionCallback(rc, taskId, subAgentId, sessionId, null); throw e instanceof RuntimeException re ? re : new RuntimeException(e); + } finally { + executionControl.executionFinished(); } } @@ -336,11 +367,15 @@ private String runRemoteTask( String taskId, String subAgentId, TaskRunSpec.RemoteTaskRunSpec remote, - boolean submitRemote) { + boolean submitRemote, + ExecutionControl executionControl) { + executionControl.executionStarted(); try { Optional latest = workspaceManager.readTaskRecord(rc, parentAgentId, sessionId, taskId); if (latest.isPresent() && latest.get().isCancelRequested()) { + confirmRemoteCancellation( + taskId, remote.baseUrl(), remote.headers(), executionControl); markCancelled(rc, sessionId, taskId); return null; } @@ -349,11 +384,19 @@ private String runRemoteTask( remote.baseUrl(), remote.headers(), taskId, subAgentId, remote.input()); updateStatus(rc, sessionId, taskId, TaskStatus.RUNNING, null, null); } - return pollRemoteUntilDone(rc, sessionId, taskId, remote.baseUrl(), remote.headers()); + return pollRemoteUntilDone( + rc, sessionId, taskId, remote.baseUrl(), remote.headers(), executionControl); } catch (Exception e) { + if (executionControl.isCancellationRequested()) { + executionControl.cancellationFailed(e); + markCancelled(rc, sessionId, taskId); + return null; + } String errMsg = e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName(); updateStatus(rc, sessionId, taskId, TaskStatus.FAILED, null, errMsg); throw e instanceof RuntimeException re ? re : new RuntimeException(e); + } finally { + executionControl.executionFinished(); } } @@ -362,18 +405,15 @@ private String pollRemoteUntilDone( String sessionId, String taskId, String baseUrl, - Map headers) + Map headers, + ExecutionControl executionControl) throws Exception { int attempt = 0; while (!Thread.currentThread().isInterrupted()) { Optional wr = workspaceManager.readTaskRecord(rc, parentAgentId, sessionId, taskId); if (wr.isPresent() && wr.get().isCancelRequested()) { - try { - protocolClient.cancelTask(baseUrl, headers, taskId); - } catch (Exception ex) { - log.debug("Remote cancel after local cancel flag: {}", ex.getMessage()); - } + confirmRemoteCancellation(taskId, baseUrl, headers, executionControl); markCancelled(rc, sessionId, taskId); return null; } @@ -406,6 +446,37 @@ private String pollRemoteUntilDone( return null; } + private void confirmRemoteCancellation( + String taskId, + String baseUrl, + Map headers, + ExecutionControl executionControl) + throws Exception { + protocolClient.cancelTask(baseUrl, headers, taskId); + int attempt = 0; + while (!Thread.currentThread().isInterrupted()) { + RemoteTaskStatus status = protocolClient.getStatus(baseUrl, headers, taskId); + String value = status.status() == null ? "" : status.status().toLowerCase(); + switch (value) { + case "cancelled", "canceled" -> { + executionControl.remoteStopConfirmed(); + return; + } + case "success", "error", "failed" -> { + executionControl.executionAlreadyStopped( + "Remote execution reached terminal status '" + value + "'"); + return; + } + default -> { + // Cancellation was accepted but the remote child is still live. + } + } + long sleepMs = Math.min(5_000L, 200L * (1L << Math.min(attempt++, 4))); + Thread.sleep(sleepMs); + } + throw new InterruptedException("Interrupted while awaiting remote cancellation"); + } + @Override public BackgroundTask getTask(RuntimeContext rc, String sessionId, String taskId) { BackgroundTask local = localTasks.get(localKey(sessionId, taskId)); @@ -449,42 +520,96 @@ public Collection listTasks( @Override public boolean cancelTask(RuntimeContext rc, String sessionId, String taskId) { - RuntimeContext effRc = rc != null ? rc : RuntimeContext.empty(); - boolean found = false; + TaskCancellation cancellation = cancelTaskWithAcknowledgement(rc, sessionId, taskId); + return cancellation.requestStatus() != TaskCancellation.RequestStatus.NOT_FOUND + && cancellation.requestStatus() != TaskCancellation.RequestStatus.REJECTED; + } - BackgroundTask local = localTasks.get(localKey(sessionId, taskId)); - if (local != null) { - local.cancel(true); - found = true; - } + @Override + public TaskCancellation cancelTaskWithAcknowledgement( + RuntimeContext rc, String sessionId, String taskId) { + RuntimeContext effRc = rc != null ? rc : RuntimeContext.empty(); + String key = localKey(sessionId, taskId); + BackgroundTask local = localTasks.get(key); + ExecutionControl control = executionControls.get(key); - // Always write cancelRequested flag to workspace for cross-node coordination Optional existing = workspaceManager.readTaskRecord(effRc, parentAgentId, sessionId, taskId); - if (existing.isPresent()) { - TaskRecord snapshot = existing.get(); - boolean agentProtocol = - snapshot.isAgentProtocolTransport() && snapshot.getRemoteBaseUrl() != null; - - TaskRecord record = snapshot; - record.setCancelRequested(true); - if (!record.getStatus().isTerminal()) { - record.setStatus(TaskStatus.CANCELLED); - } - persistRecord(effRc, sessionId, record); - - if (agentProtocol) { - try { - protocolClient.cancelTask( - snapshot.getRemoteBaseUrl(), snapshot.getRemoteHeaders(), taskId); - } catch (Exception e) { - log.warn("Remote cancel failed for task {}: {}", taskId, e.getMessage()); - } + if (existing.isEmpty() && local == null) { + TaskCancellation.ExecutionIdentity identity = + new TaskCancellation.ExecutionIdentity( + sessionId, + taskId, + null, + null, + TaskCancellation.ExecutionKind.UNKNOWN, + null); + return completedCancellation( + identity, + TaskCancellation.RequestStatus.NOT_FOUND, + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + "Task was not found"); + } + + TaskRecord snapshot = existing.orElse(null); + TaskCancellation.ExecutionIdentity identity = + control != null + ? control.identity() + : identityFromRecord(sessionId, taskId, snapshot); + TaskCancellation.RequestStatus requestStatus; + if (snapshot != null + && snapshot.getStatus() != null + && snapshot.getStatus().isTerminal() + && snapshot.getStatus() != TaskStatus.CANCELLED) { + requestStatus = TaskCancellation.RequestStatus.ALREADY_TERMINAL; + } else if (snapshot != null + && (snapshot.isCancelRequested() || snapshot.getStatus() == TaskStatus.CANCELLED)) { + requestStatus = TaskCancellation.RequestStatus.ALREADY_REQUESTED; + } else { + requestStatus = TaskCancellation.RequestStatus.ACCEPTED; + } + + if (requestStatus == TaskCancellation.RequestStatus.ALREADY_TERMINAL) { + return completedCancellation( + identity, + requestStatus, + TaskCancellation.TerminationStatus.ALREADY_STOPPED, + "Task was already in terminal state " + snapshot.getStatus()); + } + + if (snapshot != null) { + snapshot.setCancelRequested(true); + if (snapshot.getStatus() == null || !snapshot.getStatus().isTerminal()) { + snapshot.setStatus(TaskStatus.CANCELLED); } - return true; + persistRecord(effRc, sessionId, snapshot); + } + if (control != null) { + // Mark the execution control first: cancelling an adopted CompletableFuture may invoke + // its completion callback synchronously, which is not proof that the underlying work + // stopped. + control.requestCancellation(); + } + if (local != null) { + // This updates task-facing status immediately. Actual stop is acknowledged separately. + local.cancel(false); } - return found; + if (control == null && snapshot != null && snapshot.isAgentProtocolTransport()) { + control = createRemoteCancellationControl(effRc, sessionId, snapshot); + } + if (control == null) { + return completedCancellation( + identity, + requestStatus, + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + "Execution is owned by another node and no local stop handle is available"); + } + + if (!control.isCancellationRequested()) { + control.requestCancellation(); + } + return new TaskCancellation(identity, requestStatus, control.termination()); } // ---- Phase B-3 push delivery ----------------------------------------------------------- @@ -552,6 +677,7 @@ public void removeTask(RuntimeContext rc, String sessionId, String taskId) { localTasks.remove(key); localTaskSessionIds.remove(key); localTaskContexts.remove(key); + executionControls.remove(key); } @Override @@ -559,6 +685,7 @@ public void clear() { localTasks.clear(); localTaskSessionIds.clear(); localTaskContexts.clear(); + executionControls.clear(); } /** Shuts down the maintenance scheduler and (if owned) the task executor. */ @@ -725,6 +852,76 @@ void sweepOrphanedTasks(Duration orphanTimeout, Duration recentWindow) { // ---- private helpers ---- + private static TaskCancellation.ExecutionKind executionKind(TaskRunSpec spec) { + if (spec instanceof TaskRunSpec.LocalTaskRunSpec) { + return TaskCancellation.ExecutionKind.LOCAL; + } + if (spec instanceof TaskRunSpec.RemoteTaskRunSpec) { + return TaskCancellation.ExecutionKind.REMOTE; + } + if (spec instanceof TaskRunSpec.AdoptedTaskRunSpec) { + return TaskCancellation.ExecutionKind.ADOPTED; + } + return TaskCancellation.ExecutionKind.UNKNOWN; + } + + private TaskCancellation.ExecutionIdentity identityFromRecord( + String sessionId, String taskId, TaskRecord record) { + boolean remote = record != null && record.isAgentProtocolTransport(); + return new TaskCancellation.ExecutionIdentity( + sessionId, + taskId, + record != null ? record.getSubAgentId() : null, + record != null ? record.getSubSessionId() : null, + remote + ? TaskCancellation.ExecutionKind.REMOTE + : TaskCancellation.ExecutionKind.UNKNOWN, + remote ? taskId : null); + } + + private static TaskCancellation completedCancellation( + TaskCancellation.ExecutionIdentity identity, + TaskCancellation.RequestStatus requestStatus, + TaskCancellation.TerminationStatus terminationStatus, + String message) { + return new TaskCancellation( + identity, + requestStatus, + CompletableFuture.completedFuture( + new TaskCancellation.Termination(identity, terminationStatus, message))); + } + + private ExecutionControl createRemoteCancellationControl( + RuntimeContext rc, String sessionId, TaskRecord record) { + String key = localKey(sessionId, record.getTaskId()); + ExecutionControl created = + new ExecutionControl(identityFromRecord(sessionId, record.getTaskId(), record)); + ExecutionControl control = executionControls.putIfAbsent(key, created); + if (control != null) { + return control; + } + control = created; + ExecutionControl capturedControl = control; + CompletableFuture.runAsync( + () -> + runRemoteTask( + rc, + sessionId, + record.getTaskId(), + record.getSubAgentId(), + new TaskRunSpec.RemoteTaskRunSpec( + record.getRemoteBaseUrl(), + record.getRemoteHeaders() != null + ? record.getRemoteHeaders() + : Map.of(), + record.getSubAgentId(), + ""), + false, + capturedControl), + executor); + return control; + } + private static String localKey(String sessionId, String taskId) { String s = sessionId != null ? sessionId : "_"; return s + ":" + taskId; @@ -825,26 +1022,38 @@ private BackgroundTask syntheticTask(RuntimeContext rc, TaskRecord record) { k -> { CompletableFuture f = CompletableFuture.supplyAsync( - () -> - runRemoteTask( - capturedRc, - sid, - record.getTaskId(), - record.getSubAgentId(), - new TaskRunSpec - .RemoteTaskRunSpec( - record - .getRemoteBaseUrl(), - record - .getRemoteHeaders() - != null - ? record - .getRemoteHeaders() - : Map.of(), - record - .getSubAgentId(), - ""), - false), + () -> { + ExecutionControl control = + executionControls + .computeIfAbsent( + lk, + ignored -> + new ExecutionControl( + identityFromRecord( + sid, + record + .getTaskId(), + record))); + return runRemoteTask( + capturedRc, + sid, + record.getTaskId(), + record.getSubAgentId(), + new TaskRunSpec + .RemoteTaskRunSpec( + record + .getRemoteBaseUrl(), + record + .getRemoteHeaders() + != null + ? record + .getRemoteHeaders() + : Map.of(), + record.getSubAgentId(), + ""), + false, + control); + }, executor); return new BackgroundTask( record.getTaskId(), record.getSubAgentId(), f); @@ -889,4 +1098,107 @@ void onCompleted( String sessionId, String result); } + + private static final class ExecutionControl { + + private final TaskCancellation.ExecutionIdentity identity; + private final CompletableFuture termination = + new CompletableFuture<>(); + private final AtomicBoolean cancellationRequested = new AtomicBoolean(); + private final AtomicReference executionThread = new AtomicReference<>(); + + private ExecutionControl(TaskCancellation.ExecutionIdentity identity) { + this.identity = identity; + } + + TaskCancellation.ExecutionIdentity identity() { + return identity; + } + + CompletableFuture termination() { + return termination; + } + + boolean isCancellationRequested() { + return cancellationRequested.get(); + } + + void executionStarted() { + executionThread.set(Thread.currentThread()); + } + + void requestCancellation() { + cancellationRequested.set(true); + switch (identity.executionKind()) { + case LOCAL -> { + Thread thread = executionThread.get(); + if (thread != null) { + thread.interrupt(); + } + } + case ADOPTED -> + complete( + TaskCancellation.TerminationStatus.DETACHED, + "Adopted future was cancelled, but its underlying execution is not" + + " owned by the repository"); + case UNKNOWN -> + complete( + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + "No execution handle is available"); + case REMOTE -> { + // The polling execution sends the protocol cancel request and confirms status. + } + } + } + + void executionFinished() { + executionThread.set(null); + if (!cancellationRequested.get()) { + complete( + TaskCancellation.TerminationStatus.ALREADY_STOPPED, + "Execution completed before cancellation"); + return; + } + switch (identity.executionKind()) { + case LOCAL -> + complete( + TaskCancellation.TerminationStatus.COOPERATIVE_STOP_CONFIRMED, + "Local execution thread exited after cancellation"); + case REMOTE -> + complete( + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + "Remote polling ended without an authoritative terminal status"); + case ADOPTED -> + complete( + TaskCancellation.TerminationStatus.DETACHED, + "Adopted execution cannot be authoritatively observed"); + case UNKNOWN -> + complete( + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + "Execution stop cannot be observed"); + } + } + + void remoteStopConfirmed() { + complete( + TaskCancellation.TerminationStatus.COOPERATIVE_STOP_CONFIRMED, + "Remote protocol confirmed cancelled status"); + } + + void executionAlreadyStopped(String message) { + complete(TaskCancellation.TerminationStatus.ALREADY_STOPPED, message); + } + + void cancellationFailed(Throwable error) { + complete( + TaskCancellation.TerminationStatus.CANCELLATION_FAILED, + error.getMessage() != null + ? error.getMessage() + : error.getClass().getSimpleName()); + } + + private void complete(TaskCancellation.TerminationStatus status, String message) { + termination.complete(new TaskCancellation.Termination(identity, status, message)); + } + } } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/tool/TaskTool.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/tool/TaskTool.java index 8a0a245bd8..1ce9e40d3a 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/tool/TaskTool.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/tool/TaskTool.java @@ -19,6 +19,7 @@ import io.agentscope.core.tool.Tool; import io.agentscope.core.tool.ToolParam; import io.agentscope.harness.agent.subagent.task.BackgroundTask; +import io.agentscope.harness.agent.subagent.task.TaskCancellation; import io.agentscope.harness.agent.subagent.task.TaskRepository; import io.agentscope.harness.agent.subagent.task.TaskStatus; import java.time.ZoneOffset; @@ -161,8 +162,19 @@ public String taskCancel( + "\nnote: Task already in terminal state, cannot cancel."; } - taskRepository.cancelTask(runtimeContext, sessionId, taskId); - return "task_id: " + taskId + "\nstatus: cancelled\nCancellation requested successfully."; + TaskCancellation cancellation = + taskRepository.cancelTaskWithAcknowledgement(runtimeContext, sessionId, taskId); + TaskCancellation.Termination termination = + cancellation.terminationSignal().toCompletableFuture().getNow(null); + String stopStatus = + termination != null ? termination.status().name().toLowerCase() : "pending"; + return "task_id: " + + taskId + + "\nstatus: cancelled" + + "\nrequest_status: " + + cancellation.requestStatus().name().toLowerCase() + + "\nstop_status: " + + stopStatus; } @Tool( diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/subagent/task/TaskCancellationTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/subagent/task/TaskCancellationTest.java new file mode 100644 index 0000000000..aa696d1e0a --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/subagent/task/TaskCancellationTest.java @@ -0,0 +1,184 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.harness.agent.subagent.task; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agent.RuntimeContext; +import java.time.Duration; +import java.util.Collection; +import java.util.concurrent.CompletableFuture; +import org.junit.jupiter.api.Test; + +class TaskCancellationTest { + + private static final TaskCancellation.ExecutionIdentity IDENTITY = + new TaskCancellation.ExecutionIdentity( + "parent-session", + "task-1", + "subagent", + "child-session", + TaskCancellation.ExecutionKind.LOCAL, + null); + + @Test + void exposesIdentityRequestStatusAndTerminationSignal() throws Exception { + TaskCancellation.Termination expected = + new TaskCancellation.Termination( + IDENTITY, + TaskCancellation.TerminationStatus.COOPERATIVE_STOP_CONFIRMED, + "stopped"); + CompletableFuture signal = + CompletableFuture.completedFuture(expected); + TaskCancellation cancellation = + new TaskCancellation(IDENTITY, TaskCancellation.RequestStatus.ACCEPTED, signal); + + assertSame(IDENTITY, cancellation.identity()); + assertEquals(TaskCancellation.RequestStatus.ACCEPTED, cancellation.requestStatus()); + assertTrue(cancellation.isCancellationAccepted()); + assertSame(signal, cancellation.terminationSignal()); + assertSame(expected, cancellation.awaitTermination(Duration.ZERO)); + } + + @Test + void acceptanceAndStopConfirmationCoverEveryOutcome() { + TaskCancellation repeated = cancellation(TaskCancellation.RequestStatus.ALREADY_REQUESTED); + TaskCancellation rejected = cancellation(TaskCancellation.RequestStatus.REJECTED); + + assertTrue(repeated.isCancellationAccepted()); + assertFalse(rejected.isCancellationAccepted()); + assertTrue( + termination(TaskCancellation.TerminationStatus.FORCED_STOP_CONFIRMED) + .isStopConfirmed()); + assertTrue( + termination(TaskCancellation.TerminationStatus.ALREADY_STOPPED).isStopConfirmed()); + assertFalse(termination(TaskCancellation.TerminationStatus.DETACHED).isStopConfirmed()); + } + + @Test + void validatesTimeoutAndMapsExceptionalSignal() throws Exception { + TaskCancellation pending = + new TaskCancellation( + IDENTITY, + TaskCancellation.RequestStatus.ACCEPTED, + new CompletableFuture<>()); + assertEquals( + TaskCancellation.TerminationStatus.TIMED_OUT, + pending.awaitTermination(Duration.ofMillis(1)).status()); + assertThrows( + IllegalArgumentException.class, + () -> pending.awaitTermination(Duration.ofMillis(-1))); + assertThrows(NullPointerException.class, () -> pending.awaitTermination(null)); + + CompletableFuture failed = new CompletableFuture<>(); + failed.completeExceptionally(new IllegalStateException("remote cancel failed")); + TaskCancellation exceptional = + new TaskCancellation(IDENTITY, TaskCancellation.RequestStatus.ACCEPTED, failed); + TaskCancellation.Termination result = exceptional.awaitTermination(Duration.ofSeconds(1)); + assertEquals(TaskCancellation.TerminationStatus.CANCELLATION_FAILED, result.status()); + assertEquals("remote cancel failed", result.message()); + + CompletableFuture messageLessFailure = + new CompletableFuture<>(); + messageLessFailure.completeExceptionally(new RuntimeException()); + TaskCancellation noMessage = + new TaskCancellation( + IDENTITY, TaskCancellation.RequestStatus.ACCEPTED, messageLessFailure); + assertEquals( + "RuntimeException", noMessage.awaitTermination(Duration.ofSeconds(1)).message()); + } + + @Test + void legacyRepositoryDefaultAcknowledgementPreservesBooleanSemantics() throws Exception { + LegacyRepository found = new LegacyRepository(true); + LegacyRepository missing = new LegacyRepository(false); + + TaskCancellation accepted = + found.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "task-found"); + TaskCancellation notFound = + missing.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "task-missing"); + + assertEquals(TaskCancellation.RequestStatus.ACCEPTED, accepted.requestStatus()); + assertEquals("task-found", accepted.identity().taskId()); + assertEquals( + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + accepted.awaitTermination(Duration.ZERO).status()); + assertEquals(TaskCancellation.RequestStatus.NOT_FOUND, notFound.requestStatus()); + assertEquals( + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + notFound.awaitTermination(Duration.ZERO).status()); + } + + private static TaskCancellation cancellation(TaskCancellation.RequestStatus status) { + return new TaskCancellation( + IDENTITY, + status, + CompletableFuture.completedFuture( + termination(TaskCancellation.TerminationStatus.STOP_UNAVAILABLE))); + } + + private static TaskCancellation.Termination termination( + TaskCancellation.TerminationStatus status) { + return new TaskCancellation.Termination(IDENTITY, status, status.name()); + } + + private static final class LegacyRepository implements TaskRepository { + + private final boolean found; + + private LegacyRepository(boolean found) { + this.found = found; + } + + @Override + public BackgroundTask getTask(RuntimeContext rc, String sessionId, String taskId) { + return null; + } + + @Override + public BackgroundTask putTask( + RuntimeContext rc, + String taskId, + String subAgentId, + String sessionId, + TaskRunSpec spec) { + return null; + } + + @Override + public void removeTask(RuntimeContext rc, String sessionId, String taskId) {} + + @Override + public void clear() {} + + @Override + public Collection listTasks( + RuntimeContext rc, String sessionId, TaskStatus filter) { + return java.util.List.of(); + } + + @Override + public boolean cancelTask(RuntimeContext rc, String sessionId, String taskId) { + return found; + } + } +} diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/subagent/task/WorkspaceTaskCancellationTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/subagent/task/WorkspaceTaskCancellationTest.java new file mode 100644 index 0000000000..c76dbd86df --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/subagent/task/WorkspaceTaskCancellationTest.java @@ -0,0 +1,311 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.harness.agent.subagent.task; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.harness.agent.workspace.WorkspaceManager; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.time.Duration; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class WorkspaceTaskCancellationTest { + + @TempDir Path tempDir; + + private WorkspaceTaskRepository repository; + private HttpServer server; + + @BeforeEach + void setUp() { + repository = + WorkspaceTaskRepository.forTests( + new WorkspaceManager(tempDir), "cancellation-test-agent"); + } + + @AfterEach + void tearDown() { + repository.shutdown(); + if (server != null) { + server.stop(0); + } + } + + @Test + void cancelledRecordDoesNotAcknowledgeSupplierThatIsStillRunning() throws Exception { + CountDownLatch running = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicBoolean interrupted = new AtomicBoolean(); + AtomicBoolean exited = new AtomicBoolean(); + + repository.putTask( + RuntimeContext.empty(), + "task-ignores-interrupt", + "local-agent", + "session", + new TaskRunSpec.LocalTaskRunSpec( + () -> { + running.countDown(); + try { + while (release.getCount() > 0) { + try { + release.await(); + } catch (InterruptedException e) { + interrupted.set(true); + } + } + return "done"; + } finally { + exited.set(true); + } + })); + + assertTrue(running.await(5, TimeUnit.SECONDS)); + TaskCancellation cancellation = + repository.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "task-ignores-interrupt"); + + assertEquals(TaskCancellation.RequestStatus.ACCEPTED, cancellation.requestStatus()); + assertEquals(TaskCancellation.ExecutionKind.LOCAL, cancellation.identity().executionKind()); + assertEquals( + TaskCancellation.TerminationStatus.TIMED_OUT, + cancellation.awaitTermination(Duration.ofMillis(100)).status()); + assertTrue(interrupted.get(), "repository should request cooperative interruption"); + assertFalse(exited.get(), "supplier still owns execution until it actually exits"); + + release.countDown(); + TaskCancellation.Termination termination = + cancellation.awaitTermination(Duration.ofSeconds(5)); + assertEquals( + TaskCancellation.TerminationStatus.COOPERATIVE_STOP_CONFIRMED, + termination.status()); + assertTrue(termination.isStopConfirmed()); + } + + @Test + void repeatedCancellationIsIdempotentAndSharesStopAcknowledgement() throws Exception { + CountDownLatch running = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + repository.putTask( + RuntimeContext.empty(), + "task-repeat", + "local-agent", + "session", + new TaskRunSpec.LocalTaskRunSpec( + () -> { + running.countDown(); + while (release.getCount() > 0) { + try { + release.await(); + } catch (InterruptedException ignored) { + // Keep running until the test explicitly releases execution. + } + } + return "done"; + })); + + assertTrue(running.await(5, TimeUnit.SECONDS)); + TaskCancellation first = + repository.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "task-repeat"); + TaskCancellation second = + repository.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "task-repeat"); + + assertEquals(TaskCancellation.RequestStatus.ACCEPTED, first.requestStatus()); + assertEquals(TaskCancellation.RequestStatus.ALREADY_REQUESTED, second.requestStatus()); + release.countDown(); + assertEquals( + TaskCancellation.TerminationStatus.COOPERATIVE_STOP_CONFIRMED, + first.awaitTermination(Duration.ofSeconds(5)).status()); + assertEquals( + TaskCancellation.TerminationStatus.COOPERATIVE_STOP_CONFIRMED, + second.awaitTermination(Duration.ofSeconds(5)).status()); + } + + @Test + void adoptedExecutionIsReportedAsDetached() throws Exception { + CompletableFuture adopted = new CompletableFuture<>(); + repository.putTask( + RuntimeContext.empty(), + "task-adopted", + "adopted-agent", + "session", + new TaskRunSpec.AdoptedTaskRunSpec(adopted)); + + TaskCancellation cancellation = + repository.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "task-adopted"); + + assertEquals( + TaskCancellation.ExecutionKind.ADOPTED, cancellation.identity().executionKind()); + assertEquals( + TaskCancellation.TerminationStatus.DETACHED, + cancellation.awaitTermination(Duration.ofSeconds(1)).status()); + } + + @Test + void missingTaskReturnsTypedNotFoundResult() throws Exception { + TaskCancellation cancellation = + repository.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "missing"); + + assertEquals(TaskCancellation.RequestStatus.NOT_FOUND, cancellation.requestStatus()); + assertEquals( + TaskCancellation.TerminationStatus.STOP_UNAVAILABLE, + cancellation.awaitTermination(Duration.ZERO).status()); + } + + @Test + void remoteCancellationWaitsForProtocolTerminalStatus() throws Exception { + CountDownLatch submitted = new CountDownLatch(1); + AtomicReference remoteStatus = new AtomicReference<>("running"); + startTaskServer(submitted, remoteStatus, false); + + repository.putTask( + RuntimeContext.empty(), + "remote-task", + "remote-agent", + "session", + new TaskRunSpec.RemoteTaskRunSpec( + serverBaseUrl(), Map.of(), "remote-agent", "input")); + assertTrue(submitted.await(5, TimeUnit.SECONDS)); + + TaskCancellation cancellation = + repository.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "remote-task"); + TaskCancellation.Termination termination = + cancellation.awaitTermination(Duration.ofSeconds(5)); + + assertEquals( + TaskCancellation.ExecutionKind.REMOTE, cancellation.identity().executionKind()); + assertEquals("remote-task", cancellation.identity().remoteTaskId()); + assertEquals( + TaskCancellation.TerminationStatus.COOPERATIVE_STOP_CONFIRMED, + termination.status()); + } + + @Test + void remoteProtocolFailureIsAnAuthoritativeCancellationResult() throws Exception { + CountDownLatch submitted = new CountDownLatch(1); + startTaskServer(submitted, new AtomicReference<>("running"), true); + + repository.putTask( + RuntimeContext.empty(), + "remote-failure", + "remote-agent", + "session", + new TaskRunSpec.RemoteTaskRunSpec( + serverBaseUrl(), Map.of(), "remote-agent", "input")); + assertTrue(submitted.await(5, TimeUnit.SECONDS)); + + TaskCancellation cancellation = + repository.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "remote-failure"); + + assertEquals( + TaskCancellation.TerminationStatus.CANCELLATION_FAILED, + cancellation.awaitTermination(Duration.ofSeconds(5)).status()); + } + + @Test + void completedTaskReturnsAlreadyTerminalAcknowledgement() throws Exception { + repository.putTask( + RuntimeContext.empty(), + "already-complete", + "local-agent", + "session", + new TaskRunSpec.LocalTaskRunSpec(() -> "done")); + awaitTaskStatus("session", "already-complete", TaskStatus.COMPLETED); + + TaskCancellation cancellation = + repository.cancelTaskWithAcknowledgement( + RuntimeContext.empty(), "session", "already-complete"); + + assertEquals(TaskCancellation.RequestStatus.ALREADY_TERMINAL, cancellation.requestStatus()); + assertEquals( + TaskCancellation.TerminationStatus.ALREADY_STOPPED, + cancellation.awaitTermination(Duration.ZERO).status()); + } + + private void awaitTaskStatus(String sessionId, String taskId, TaskStatus expected) + throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (System.nanoTime() < deadline) { + BackgroundTask task = repository.getTask(RuntimeContext.empty(), sessionId, taskId); + if (task != null && task.getTaskStatus() == expected) { + return; + } + Thread.sleep(10); + } + throw new AssertionError("Task did not reach status " + expected); + } + + private void startTaskServer( + CountDownLatch submitted, + AtomicReference remoteStatus, + boolean failCancellation) + throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/tasks", + exchange -> { + String path = exchange.getRequestURI().getPath(); + if ("/tasks".equals(path)) { + submitted.countDown(); + respond(exchange, 202, "{}"); + } else if (path.endsWith("/cancel")) { + if (failCancellation) { + respond(exchange, 500, "cancel failed"); + } else { + remoteStatus.set("cancelled"); + respond(exchange, 200, "{}"); + } + } else { + respond(exchange, 200, "{\"status\":\"" + remoteStatus.get() + "\"}"); + } + }); + server.start(); + } + + private String serverBaseUrl() { + return "http://127.0.0.1:" + server.getAddress().getPort(); + } + + private static void respond(HttpExchange exchange, int status, String body) throws IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(status, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + } +} diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/TaskToolCancellationTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/TaskToolCancellationTest.java new file mode 100644 index 0000000000..cea25c5664 --- /dev/null +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/tool/TaskToolCancellationTest.java @@ -0,0 +1,117 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.harness.agent.tool; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.harness.agent.subagent.task.BackgroundTask; +import io.agentscope.harness.agent.subagent.task.TaskCancellation; +import io.agentscope.harness.agent.subagent.task.TaskRepository; +import io.agentscope.harness.agent.subagent.task.TaskRunSpec; +import io.agentscope.harness.agent.subagent.task.TaskStatus; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import org.junit.jupiter.api.Test; + +class TaskToolCancellationTest { + + @Test + void taskCancelReportsImmediateStopAcknowledgement() { + CompletableFuture signal = new CompletableFuture<>(); + StubRepository repository = new StubRepository(signal); + signal.complete( + new TaskCancellation.Termination( + repository.identity, + TaskCancellation.TerminationStatus.COOPERATIVE_STOP_CONFIRMED, + "stopped")); + + String output = + new TaskTool(repository) + .taskCancel( + RuntimeContext.builder().sessionId("session").build(), "task-1"); + + assertTrue(output.contains("request_status: accepted")); + assertTrue(output.contains("stop_status: cooperative_stop_confirmed")); + } + + @Test + void taskCancelReportsPendingWhenExecutionHasNotStopped() { + StubRepository repository = new StubRepository(new CompletableFuture<>()); + + String output = new TaskTool(repository).taskCancel(RuntimeContext.empty(), "task-1"); + + assertTrue(output.contains("request_status: accepted")); + assertTrue(output.contains("stop_status: pending")); + } + + private static final class StubRepository implements TaskRepository { + + private final CompletableFuture signal; + private final CompletableFuture taskFuture = new CompletableFuture<>(); + private final TaskCancellation.ExecutionIdentity identity = + new TaskCancellation.ExecutionIdentity( + "session", + "task-1", + "agent", + null, + TaskCancellation.ExecutionKind.LOCAL, + null); + + private StubRepository(CompletableFuture signal) { + this.signal = signal; + } + + @Override + public BackgroundTask getTask(RuntimeContext rc, String sessionId, String taskId) { + return new BackgroundTask(taskId, "agent", taskFuture); + } + + @Override + public TaskCancellation cancelTaskWithAcknowledgement( + RuntimeContext rc, String sessionId, String taskId) { + return new TaskCancellation(identity, TaskCancellation.RequestStatus.ACCEPTED, signal); + } + + @Override + public BackgroundTask putTask( + RuntimeContext rc, + String taskId, + String subAgentId, + String sessionId, + TaskRunSpec spec) { + return null; + } + + @Override + public void removeTask(RuntimeContext rc, String sessionId, String taskId) {} + + @Override + public void clear() {} + + @Override + public Collection listTasks( + RuntimeContext rc, String sessionId, TaskStatus filter) { + return List.of(); + } + + @Override + public boolean cancelTask(RuntimeContext rc, String sessionId, String taskId) { + return true; + } + } +}