From 7fbdc37329d891bcdceda44f0ced205d5328dc07 Mon Sep 17 00:00:00 2001 From: "ark-hand[bot]" Date: Fri, 11 Sep 2026 03:06:28 +0000 Subject: [PATCH] fix(selfHosted): filter recovered tool results by session blockers Sync-Source-Commit: 3d1df8fbeb78a4093bf8a337f8dda94fc1919fd9 Hand-Written-Reason: No Ark-APIs provenance marker; treated as a hand-written source commit. Release-Version: 0.7.0 --- .../ark/runtime/selfhosted/Event.java | 19 ++ .../selfhosted/FileToolResultStore.java | 8 + .../selfhosted/SelfHostedConstants.java | 3 + .../runtime/selfhosted/SessionToolRunner.java | 88 +++++++++- .../selfhosted/FileToolResultStoreTest.java | 16 ++ .../selfhosted/SessionToolRunnerTest.java | 165 ++++++++++++++++++ 6 files changed, 296 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/volcengine/ark/runtime/selfhosted/Event.java b/src/main/java/com/volcengine/ark/runtime/selfhosted/Event.java index b0ff6bc..0a97e89 100644 --- a/src/main/java/com/volcengine/ark/runtime/selfhosted/Event.java +++ b/src/main/java/com/volcengine/ark/runtime/selfhosted/Event.java @@ -138,6 +138,25 @@ public String stopReasonType() { return stopReason instanceof String ? (String) stopReason : ""; } + @SuppressWarnings("unchecked") + public List stopReasonEventIds() { + if (!(stopReason instanceof Map)) { + return new ArrayList<>(); + } + Object rawIds = ((Map) stopReason).get("event_ids"); + if (!(rawIds instanceof List)) { + return new ArrayList<>(); + } + List ids = new ArrayList<>(); + for (Object rawId : (List) rawIds) { + String id = stringValue(rawId); + if (!id.isEmpty()) { + ids.add(id); + } + } + return ids; + } + public String callId() { if (toolUseId != null && !toolUseId.isEmpty()) { return toolUseId; diff --git a/src/main/java/com/volcengine/ark/runtime/selfhosted/FileToolResultStore.java b/src/main/java/com/volcengine/ark/runtime/selfhosted/FileToolResultStore.java index a2e1df9..7495faf 100644 --- a/src/main/java/com/volcengine/ark/runtime/selfhosted/FileToolResultStore.java +++ b/src/main/java/com/volcengine/ark/runtime/selfhosted/FileToolResultStore.java @@ -122,6 +122,14 @@ public void markSent(String callId) throws IOException { write(record); } + public void discard(String callId) throws IOException { + if (callId == null || callId.isEmpty()) { + throw new IllegalArgumentException("call id must not be empty"); + } + Files.deleteIfExists(path(callId)); + syncDirectory(); + } + private Map read(String callId) throws IOException { Path path = path(callId); if (!Files.exists(path)) { diff --git a/src/main/java/com/volcengine/ark/runtime/selfhosted/SelfHostedConstants.java b/src/main/java/com/volcengine/ark/runtime/selfhosted/SelfHostedConstants.java index 114f791..8f1c8f2 100644 --- a/src/main/java/com/volcengine/ark/runtime/selfhosted/SelfHostedConstants.java +++ b/src/main/java/com/volcengine/ark/runtime/selfhosted/SelfHostedConstants.java @@ -13,6 +13,8 @@ public final class SelfHostedConstants { public static final String EVENT_TYPE_USER_TOOL_RESULT = "user.tool_result"; public static final String EVENT_TYPE_USER_CUSTOM_TOOL_RESULT = "user.custom_tool_result"; public static final String EVENT_TYPE_SESSION_STATUS_IDLE = "session.status_idle"; + public static final String EVENT_TYPE_SESSION_STATUS_RUNNING = "session.status_running"; + public static final String EVENT_TYPE_SESSION_STATUS_RESCHEDULED = "session.status_rescheduled"; public static final String EVENT_TYPE_SESSION_STATUS_TERMINATED = "session.status_terminated"; public static final String EVENT_TYPE_SESSION_DELETED = "session.deleted"; @@ -25,6 +27,7 @@ public final class SelfHostedConstants { public static final String EVENT_LIST_ORDER_ASC = "asc"; public static final String SESSION_STOP_REASON_END_TURN = "end_turn"; + public static final String SESSION_STOP_REASON_REQUIRES_ACTION = "requires_action"; public static final long DEFAULT_MAX_IDLE_MILLIS = 60000L; public static final long DEFAULT_TOOL_TIMEOUT_MILLIS = 120000L; diff --git a/src/main/java/com/volcengine/ark/runtime/selfhosted/SessionToolRunner.java b/src/main/java/com/volcengine/ark/runtime/selfhosted/SessionToolRunner.java index eb1b535..70faeab 100644 --- a/src/main/java/com/volcengine/ark/runtime/selfhosted/SessionToolRunner.java +++ b/src/main/java/com/volcengine/ark/runtime/selfhosted/SessionToolRunner.java @@ -66,6 +66,9 @@ public List run() throws IOException { if (options.resultStore != null) { FileToolResultStore.RecoverResult recovered = options.resultStore.recover(); state.pendingResults.putAll(recovered.getPending()); + for (String callId : recovered.getPending().keySet()) { + state.recoveredResults.put(callId, Boolean.TRUE); + } state.processed.putAll(recovered.getProcessed()); state.answered.putAll(recovered.getProcessed()); } @@ -202,8 +205,8 @@ private void reconcileOnce(boolean reconcile) throws IOException { private void consumeList() throws IOException { while (!isClosed()) { - flushResults(); reconcile(false); + flushResults(); if (idleExpired()) { throw new IdleTimeoutException(); } @@ -231,6 +234,7 @@ private void processListedEvents(List events, boolean reconcile) throws I if (!reconcile && !seenNow) { continue; } + observeSessionState(event); if (seenNow && !SelfHostedConstants.EVENT_TYPE_USER_TOOL_CONFIRMATION.equals(event.getType())) { touchedIdle = true; lastWasEndTurn = SelfHostedConstants.EVENT_TYPE_SESSION_STATUS_IDLE.equals(event.getType()) @@ -254,11 +258,12 @@ private void processListedEvents(List events, boolean reconcile) throws I throw new SessionTerminatedException(); } } + reconcileRecoveredResults(); if (touchedIdle) { disarmIdle(); } for (Event event : pending) { - if (!isAnswered(event.callId())) { + if (!isAnswered(event.callId()) && shouldHandleToolUse(event.callId())) { handleToolUse(event, SelfHostedConstants.EVENT_TYPE_AGENT_CUSTOM_TOOL_USE.equals(event.getType())); } } @@ -288,6 +293,8 @@ private void handleStreamEvent(Event event) throws IOException { if (!markEventSeen(event)) { return; } + observeSessionState(event); + reconcileRecoveredResults(); noteIdleEvent(event); handleEvent(event); } @@ -316,6 +323,9 @@ private void handleToolUse(Event event, boolean custom) throws IOException { } Event pending = state.pendingResults.get(callId); if (pending != null) { + if (state.recoveredResults.containsKey(callId)) { + return; + } sendResult(callId, event, custom, "", pending); return; } @@ -468,6 +478,9 @@ private boolean retrySendEvent(Event event) { private void flushResults() throws IOException { for (Map.Entry entry : new ArrayList<>(state.pendingResults.entrySet())) { + if (state.recoveredResults.containsKey(entry.getKey())) { + continue; + } if (retrySendEvent(entry.getValue())) { markAnswered(entry.getKey()); if (options.resultStore != null) { @@ -486,6 +499,69 @@ private void flushResults() throws IOException { maybeArmPendingIdle(); } + private void observeSessionState(Event event) { + String type = event.getType(); + if (SelfHostedConstants.EVENT_TYPE_AGENT_TOOL_USE.equals(type) + || SelfHostedConstants.EVENT_TYPE_AGENT_CUSTOM_TOOL_USE.equals(type)) { + String callId = event.callId(); + if (!callId.isEmpty()) { + state.sessionToolUses.put(callId, Boolean.TRUE); + state.toolUsesSinceStatus.put(callId, Boolean.TRUE); + } + return; + } + if (SelfHostedConstants.EVENT_TYPE_SESSION_STATUS_IDLE.equals(type)) { + state.blockingEventsKnown = true; + state.blockingEventIds.clear(); + if (SelfHostedConstants.SESSION_STOP_REASON_REQUIRES_ACTION.equals(event.stopReasonType())) { + for (String eventId : event.stopReasonEventIds()) { + state.blockingEventIds.put(eventId, Boolean.TRUE); + } + } + state.toolUsesSinceStatus.clear(); + return; + } + if (SelfHostedConstants.EVENT_TYPE_SESSION_STATUS_RUNNING.equals(type) + || SelfHostedConstants.EVENT_TYPE_SESSION_STATUS_RESCHEDULED.equals(type)) { + state.blockingEventsKnown = true; + state.blockingEventIds.clear(); + state.toolUsesSinceStatus.clear(); + } + } + + private boolean shouldHandleToolUse(String callId) { + if (!state.blockingEventsKnown) { + return true; + } + return state.blockingEventIds.containsKey(callId) || state.toolUsesSinceStatus.containsKey(callId); + } + + private void reconcileRecoveredResults() { + if (!state.blockingEventsKnown) { + return; + } + for (String callId : new ArrayList<>(state.recoveredResults.keySet())) { + if (state.blockingEventIds.containsKey(callId) && state.sessionToolUses.containsKey(callId)) { + state.recoveredResults.remove(callId); + continue; + } + if (state.toolUsesSinceStatus.containsKey(callId)) { + continue; + } + state.recoveredResults.remove(callId); + state.pendingResults.remove(callId); + LOGGER.warning("discard stale recovered tool result tool_use_id=" + callId); + if (options.resultStore != null) { + try { + options.resultStore.discard(callId); + } catch (IOException error) { + LOGGER.log(Level.WARNING, "discard persisted tool result failed tool_use_id=" + callId, error); + } + } + } + maybeArmPendingIdle(); + } + private boolean ownsTool(Event event, boolean custom) { return custom ? options.customTools.containsKey(event.getName()) : options.tools.has(event.getName()); } @@ -537,6 +613,7 @@ private void markAnswered(String callId) { state.answered.put(callId, Boolean.TRUE); state.processed.put(callId, Boolean.TRUE); state.pendingResults.remove(callId); + state.recoveredResults.remove(callId); state.pendingAsk.remove(callId); state.externalTools.remove(callId); maybeArmPendingIdle(); @@ -565,7 +642,7 @@ private void releaseConfirmedToolUses() throws IOException { private boolean hasUnblockedOutstandingTool(List pending) { for (Event event : pending) { String callId = event.callId(); - if (callId.isEmpty() || isAnswered(callId)) { + if (callId.isEmpty() || isAnswered(callId) || !shouldHandleToolUse(callId)) { continue; } if (state.pendingAsk.containsKey(callId) || state.pendingResults.containsKey(callId)) { @@ -661,9 +738,14 @@ private static class State { Map seen = new LinkedHashMap<>(); Map answered = new LinkedHashMap<>(); Map pendingResults = new LinkedHashMap<>(); + Map recoveredResults = new LinkedHashMap<>(); Map pendingAsk = new LinkedHashMap<>(); Map confirmations = new LinkedHashMap<>(); Map externalTools = new LinkedHashMap<>(); + Map sessionToolUses = new LinkedHashMap<>(); + Map toolUsesSinceStatus = new LinkedHashMap<>(); + Map blockingEventIds = new LinkedHashMap<>(); + boolean blockingEventsKnown; long idleArmedAt; boolean idleArmPending; } diff --git a/src/test/java/com/volcengine/ark/runtime/selfhosted/FileToolResultStoreTest.java b/src/test/java/com/volcengine/ark/runtime/selfhosted/FileToolResultStoreTest.java index 0da1be3..c952120 100644 --- a/src/test/java/com/volcengine/ark/runtime/selfhosted/FileToolResultStoreTest.java +++ b/src/test/java/com/volcengine/ark/runtime/selfhosted/FileToolResultStoreTest.java @@ -73,4 +73,20 @@ public void sessionStoreSanitizesSessionId() throws Exception { } assertFalse(Files.exists(workdir.resolve("outside"))); } + + @Test + public void discardRemovesRecoveredRecord() throws Exception { + Path workdir = Files.createTempDirectory("ark-java-store-"); + FileToolResultStore store = new FileToolResultStore(workdir.toString(), "session-a"); + Map raw = new LinkedHashMap<>(); + raw.put("id", "call-1"); + raw.put("type", "agent.tool_use"); + raw.put("name", "bash"); + store.begin("call-1", Event.fromMap(raw)); + + store.discard("call-1"); + + assertTrue(store.recover().getPending().isEmpty()); + assertTrue(store.recover().getProcessed().isEmpty()); + } } diff --git a/src/test/java/com/volcengine/ark/runtime/selfhosted/SessionToolRunnerTest.java b/src/test/java/com/volcengine/ark/runtime/selfhosted/SessionToolRunnerTest.java index c199e98..6a56549 100644 --- a/src/test/java/com/volcengine/ark/runtime/selfhosted/SessionToolRunnerTest.java +++ b/src/test/java/com/volcengine/ark/runtime/selfhosted/SessionToolRunnerTest.java @@ -203,6 +203,77 @@ public void customToolTimeoutAbandonsNoncooperativeTool() throws Exception { assertToolTimeoutAbandonsNoncooperativeTool(true); } + @Test + public void recoveredResultsAreFilteredAgainstCurrentBlockers() throws Exception { + List discarded = new ArrayList<>(); + List marked = new ArrayList<>(); + FileToolResultStore store = recordingStore(discarded, marked); + AtomicInteger executions = new AtomicInteger(); + List sent = new ArrayList<>(); + Tool tool = countingTool(executions); + SessionToolRunner runner = runnerWithStore(store, tool, sent); + stateMap(runner, "pendingResults").put( + "foreign-call", Event.newUserCustomToolResultEvent("foreign-call", Collections.emptyList(), false, "")); + stateMap(runner, "pendingResults").put( + "stale-call", Event.newUserCustomToolResultEvent("stale-call", Collections.emptyList(), false, "")); + stateMap(runner, "recoveredResults").put("foreign-call", Boolean.TRUE); + stateMap(runner, "recoveredResults").put("stale-call", Boolean.TRUE); + + processListedEvents( + runner, + java.util.Arrays.asList( + toolUse("stale-call"), + toolUse("current-call"), + requiresAction("current-call"))); + + assertEquals(1, executions.get()); + assertEquals(1, sent.size()); + assertEquals("current-call", sent.get(0).resultCallId()); + assertTrue(discarded.contains("foreign-call")); + assertTrue(discarded.contains("stale-call")); + assertTrue(stateMap(runner, "pendingResults").isEmpty()); + assertTrue(stateMap(runner, "recoveredResults").isEmpty()); + runner.close(); + } + + @Test + public void currentBlockerReusesRecoveredResultWithoutReexecution() throws Exception { + List discarded = new ArrayList<>(); + List marked = new ArrayList<>(); + FileToolResultStore store = recordingStore(discarded, marked); + AtomicInteger executions = new AtomicInteger(); + List sent = new ArrayList<>(); + SessionToolRunner runner = runnerWithStore(store, countingTool(executions), sent); + Event result = Event.newUserCustomToolResultEvent("current-call", Collections.emptyList(), false, ""); + stateMap(runner, "pendingResults").put("current-call", result); + stateMap(runner, "recoveredResults").put("current-call", Boolean.TRUE); + + processListedEvents(runner, java.util.Arrays.asList(toolUse("current-call"), requiresAction("current-call"))); + + assertEquals(0, executions.get()); + assertEquals(Collections.singletonList(result), sent); + assertEquals(Collections.singletonList("current-call"), marked); + assertTrue(discarded.isEmpty()); + runner.close(); + } + + @Test + public void recoveredResultWaitsForAuthoritativeStatus() throws Exception { + AtomicInteger executions = new AtomicInteger(); + List sent = new ArrayList<>(); + SessionToolRunner runner = runnerWithStore(null, countingTool(executions), sent); + stateMap(runner, "pendingResults").put( + "current-call", Event.newUserCustomToolResultEvent("current-call", Collections.emptyList(), false, "")); + stateMap(runner, "recoveredResults").put("current-call", Boolean.TRUE); + + processListedEvents(runner, Collections.singletonList(toolUse("current-call"))); + + assertEquals(0, executions.get()); + assertTrue(sent.isEmpty()); + assertTrue(stateMap(runner, "recoveredResults").containsKey("current-call")); + runner.close(); + } + private static void assertToolTimeoutAbandonsNoncooperativeTool(boolean custom) throws Exception { CountDownLatch started = new CountDownLatch(1); CountDownLatch release = new CountDownLatch(1); @@ -269,6 +340,90 @@ private static SessionToolRunner idleRunner() throws IOException { Files.createTempDirectory("ark-java-idle-").toString()))); } + private static FileToolResultStore recordingStore(List discarded, List marked) + throws IOException { + return new FileToolResultStore(Files.createTempDirectory("ark-java-recovery-store-").toString()) { + @Override + public ToolCallStoreDecision begin(String callId, Event event) { + return new ToolCallStoreDecision(false, null); + } + + @Override + public void saveResult(String callId, Event result) { + } + + @Override + public void markSent(String callId) { + marked.add(callId); + } + + @Override + public void discard(String callId) { + discarded.add(callId); + } + }; + } + + private static Tool countingTool(AtomicInteger executions) { + return new Tool() { + @Override + public String name() { + return "custom"; + } + + @Override + public ToolResult execute(Object input, ToolContext context) { + executions.incrementAndGet(); + return ToolResult.text("ok"); + } + }; + } + + private static SessionToolRunner runnerWithStore(FileToolResultStore store, Tool tool, List sent) + throws IOException { + SelfHostedClient client = new SelfHostedClient("test-key") { + @Override + public void sendEvent(String sessionId, Event event) { + sent.add(event); + } + }; + return new SessionToolRunner( + client, + "session-1", + new SessionToolRunner.Options() + .tools(new ToolSet()) + .toolContext(new ToolContext( + Files.createTempDirectory("ark-java-recovery-runner-").toString())) + .customTools(Collections.singletonMap("custom", tool)) + .resultStore(store)); + } + + private static Event toolUse(String callId) { + Map raw = new LinkedHashMap<>(); + raw.put("id", callId); + raw.put("type", "agent.custom_tool_use"); + raw.put("name", "custom"); + raw.put("input", Collections.emptyMap()); + return Event.fromMap(raw); + } + + private static Event requiresAction(String callId) { + Map stopReason = new LinkedHashMap<>(); + stopReason.put("type", "requires_action"); + stopReason.put("event_ids", Collections.singletonList(callId)); + Map raw = new LinkedHashMap<>(); + raw.put("id", "idle-" + callId); + raw.put("type", "session.status_idle"); + raw.put("stop_reason", stopReason); + return Event.fromMap(raw); + } + + private static void processListedEvents(SessionToolRunner runner, List events) throws Exception { + Method process = SessionToolRunner.class.getDeclaredMethod("processListedEvents", List.class, boolean.class); + process.setAccessible(true); + process.invoke(runner, events, true); + } + private static Event idleEvent() { Map stopReason = new LinkedHashMap<>(); stopReason.put("type", "end_turn"); @@ -298,6 +453,16 @@ private static Map answered(SessionToolRunner runner) throws Ex return (Map) answeredField.get(state); } + @SuppressWarnings("unchecked") + private static Map stateMap(SessionToolRunner runner, String fieldName) throws Exception { + Field stateField = SessionToolRunner.class.getDeclaredField("state"); + stateField.setAccessible(true); + Object state = stateField.get(runner); + Field field = state.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + return (Map) field.get(state); + } + private static Response response(Request request, int code, String body) throws IOException { return new Response.Builder() .request(request)