From 436bd3be01dda4ca4f71e4cd69729142c5d768bb Mon Sep 17 00:00:00 2001 From: dargoner <1793850+dargoner@users.noreply.github.com> Date: Thu, 29 Oct 2026 19:41:10 +0800 Subject: [PATCH] fix(harness): redact media in prepared-request logs and chain the first overflow cause --- .../core/session/SessionRecorder.java | 101 ++++++++- .../core/session/ReActSessionLogTest.java | 118 ++++++++++ .../middleware/CompactionMiddleware.java | 55 ++++- .../middleware/CompactionMiddlewareTest.java | 209 ++++++++++++++++++ docs/v2/en/docs/harness/session-log.md | 2 + docs/v2/zh/docs/harness/session-log.md | 2 + 6 files changed, 478 insertions(+), 9 deletions(-) diff --git a/agentscope-core/src/main/java/io/agentscope/core/session/SessionRecorder.java b/agentscope-core/src/main/java/io/agentscope/core/session/SessionRecorder.java index 7425582283..df0ddaa594 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/session/SessionRecorder.java +++ b/agentscope-core/src/main/java/io/agentscope/core/session/SessionRecorder.java @@ -22,12 +22,19 @@ import io.agentscope.core.event.RequireExternalExecutionEvent; import io.agentscope.core.event.RequireUserConfirmEvent; import io.agentscope.core.event.UserConfirmResultEvent; +import io.agentscope.core.message.AudioBlock; +import io.agentscope.core.message.Base64Source; import io.agentscope.core.message.ContentBlock; +import io.agentscope.core.message.DataBlock; import io.agentscope.core.message.GenerateReason; +import io.agentscope.core.message.ImageBlock; import io.agentscope.core.message.Msg; import io.agentscope.core.message.MsgRole; +import io.agentscope.core.message.Source; import io.agentscope.core.message.ToolResultBlock; import io.agentscope.core.message.ToolUseBlock; +import io.agentscope.core.message.URLSource; +import io.agentscope.core.message.VideoBlock; import io.agentscope.core.middleware.ModelCallInput; import io.agentscope.core.model.ChatResponse; import io.agentscope.core.state.AgentState; @@ -599,7 +606,7 @@ public Mono prepareModel( payload.put("modelCallId", callId); payload.put("purpose", purpose); payload.put("model", input.model().getModelName()); - payload.put("messages", input.messages()); + payload.put("messages", redactedForLog(input.messages())); payload.put("tools", input.tools()); payload.put("options", input.options()); append("request/prepared", payload); @@ -610,4 +617,96 @@ public Mono prepareModel( return flush(); }); } + + /** + * Placeholder URL used in the logged request projection. Media bytes (Base64) and signed or + * temporary URLs are stripped before the request is written to the session log; the block type + * is preserved so the logged shape still reflects one image/audio/video/data payload. + */ + private static final String REDACTED_MEDIA_URL = "redacted"; + + /** + * Media-safe projection of the prepared request for the session log. + * + *

The canonical history keeps only attachment placeholders; the model boundary is where + * references are expanded into real media. Writing that expansion verbatim would persist Base64 + * bodies or signed URLs, so every media block is replaced by a source with no payload. Text, + * roles, block order and non-media metadata are untouched; the request sent to the model is + * unmodified because only this copy is logged. + */ + private static List redactedForLog(List messages) { + if (messages == null) return List.of(); + List redacted = new ArrayList<>(messages.size()); + for (Msg message : messages) redacted.add(redactMessage(message)); + return redacted; + } + + private static Msg redactMessage(Msg message) { + List content = message.getContent(); + if (content == null || content.stream().noneMatch(SessionRecorder::containsMedia)) + return message; + List redacted = new ArrayList<>(content.size()); + for (ContentBlock block : content) redacted.add(redactMediaBlock(block)); + return Msg.builder() + .id(message.getId()) + .name(message.getName()) + .role(message.getRole()) + .content(redacted) + .metadata(message.getMetadata()) + .timestamp(message.getTimestamp()) + .usage(message.getUsage()) + .build(); + } + + private static boolean containsMedia(ContentBlock block) { + if (block instanceof ImageBlock + || block instanceof AudioBlock + || block instanceof VideoBlock + || block instanceof DataBlock) { + return true; + } + if (block instanceof ToolResultBlock result) { + List output = result.getOutput(); + return output != null && output.stream().anyMatch(SessionRecorder::containsMedia); + } + return false; + } + + private static ContentBlock redactMediaBlock(ContentBlock block) { + if (block instanceof ImageBlock image) + return new ImageBlock(redactedSource(image.getSource())); + if (block instanceof AudioBlock audio) + return new AudioBlock(redactedSource(audio.getSource())); + if (block instanceof VideoBlock video) + return new VideoBlock(redactedSource(video.getSource())); + if (block instanceof DataBlock data) + return DataBlock.builder() + .source(redactedSource(data.getSource())) + .id(data.getId()) + .name(data.getName()) + .build(); + if (block instanceof ToolResultBlock result) { + List output = result.getOutput(); + if (output == null || output.stream().noneMatch(SessionRecorder::containsMedia)) + return result; + List redactedOutput = new ArrayList<>(output.size()); + for (ContentBlock child : output) redactedOutput.add(redactMediaBlock(child)); + return ToolResultBlock.builder() + .id(result.getId()) + .name(result.getName()) + .output(redactedOutput) + .metadata(result.getMetadata()) + .state(result.getState()) + .executionDetails(result.getExecutionDetails()) + .build(); + } + return block; + } + + private static Source redactedSource(Source source) { + String mediaType = null; + if (source instanceof Base64Source base64) mediaType = base64.getMediaType(); + else if (source instanceof URLSource url) mediaType = url.getMimeType(); + return new URLSource(REDACTED_MEDIA_URL, mediaType); + } } diff --git a/agentscope-core/src/test/java/io/agentscope/core/session/ReActSessionLogTest.java b/agentscope-core/src/test/java/io/agentscope/core/session/ReActSessionLogTest.java index 7998c87f27..4f7ef8d13c 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/session/ReActSessionLogTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/session/ReActSessionLogTest.java @@ -16,14 +16,21 @@ package io.agentscope.core.session; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import io.agentscope.core.ReActAgent; import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.core.message.Base64Source; +import io.agentscope.core.message.ImageBlock; import io.agentscope.core.message.Msg; +import io.agentscope.core.message.MsgRole; import io.agentscope.core.message.TextBlock; +import io.agentscope.core.message.ToolResultBlock; +import io.agentscope.core.message.URLSource; import io.agentscope.core.message.UserMessage; +import io.agentscope.core.middleware.ModelCallInput; import io.agentscope.core.model.ChatResponse; import io.agentscope.core.model.GenerateOptions; import io.agentscope.core.model.Model; @@ -84,6 +91,117 @@ void rebuiltAgentRestoresCompleteHistoryAndRecordsFinalRequest() { .isEmpty()); } + @Test + void preparedRequestLogRedactsMediaButKeepsTextAndModelReceivesRealPayload() { + var store = new InMemorySessionLogStore(); + var model = new CaptureModel(); + var agent = ReActAgent.builder().name("media").model(model).sessionLogStore(store).build(); + String base64 = "data:image/png;base64,SECRETBASE64PAYLOAD"; + String signedUrl = "https://cdn.example.test/img.png?token=SECRETTOKEN"; + agent.call( + List.of( + new UserMessage( + TextBlock.builder().text("hello").build(), + ImageBlock.builder() + .source( + Base64Source.builder() + .mediaType("image/png") + .data(base64) + .build()) + .build(), + ImageBlock.builder() + .source(new URLSource(signedUrl, "image/png")) + .build())), + context()) + .block(Duration.ofSeconds(10)); + + // The model still receives the real media bytes and URL. + Msg sent = model.calls.get(0).get(0); + assertTrue(sent.getContentBlocks(ImageBlock.class).size() == 2); + assertTrue( + sent.getContentBlocks(ImageBlock.class).stream() + .anyMatch( + image -> + image.getSource() instanceof Base64Source source + && base64.equals(source.getData()))); + assertTrue( + sent.getContentBlocks(ImageBlock.class).stream() + .anyMatch( + image -> + image.getSource() instanceof URLSource source + && signedUrl.equals(source.getUrl()))); + + var events = store.open(new SessionKey("u", "media", "s"), context()).readAfter(0, 1000); + var prepared = + events.stream() + .filter(e -> e.type().equals("request/prepared")) + .findFirst() + .orElseThrow(); + String payload = prepared.payloadJson(); + assertFalse(payload.contains("SECRETBASE64PAYLOAD"), payload); + assertFalse(payload.contains("SECRETTOKEN"), payload); + assertTrue(payload.contains("hello")); + assertTrue(payload.contains("redacted")); + // Block types survive the projection so the logged request still shows two images. + assertTrue(payload.contains("image"), payload); + } + + @Test + void preparedRequestLogRedactsMediaNestedInToolResults() { + var store = new InMemorySessionLogStore(); + SessionLog log = store.open(new SessionKey("u", "tool", "s"), context()); + var recorder = new SessionRecorder(log, "turn-1"); + String base64 = "SECRETTOOLRESULTBASE64"; + Msg toolMessage = + Msg.builder() + .role(MsgRole.TOOL) + .content( + ToolResultBlock.builder() + .id("call-1") + .name("screenshot") + .output( + List.of( + TextBlock.builder().text("ok").build(), + ImageBlock.builder() + .source( + Base64Source.builder() + .mediaType( + "image/png") + .data(base64) + .build()) + .build())) + .build()) + .build(); + Model model = + new Model() { + public String getModelName() { + return "fake"; + } + + public Flux stream( + List messages, List tools, GenerateOptions options) { + return Flux.empty(); + } + }; + + recorder.prepareModel( + new ModelCallInput(List.of(toolMessage), List.of(), null, model), + "call-1", + "REASONING", + AgentState.builder().sessionId("s").context(List.of()).build()) + .block(Duration.ofSeconds(5)); + + var events = store.open(new SessionKey("u", "tool", "s"), context()).readAfter(0, 100); + var prepared = + events.stream() + .filter(e -> e.type().equals("request/prepared")) + .findFirst() + .orElseThrow(); + assertFalse(prepared.payloadJson().contains(base64), prepared.payloadJson()); + assertTrue(prepared.payloadJson().contains("redacted")); + assertTrue(prepared.payloadJson().contains("ok")); + } + @Test void nativeAdministrationUsesFencedCheckpointsAndNeverLoadsLegacyState() { var store = new InMemorySessionLogStore(); diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/CompactionMiddleware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/CompactionMiddleware.java index b441768e90..490b39e50e 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/CompactionMiddleware.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/CompactionMiddleware.java @@ -44,10 +44,12 @@ import java.util.Locale; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CancellationException; import java.util.function.Function; import java.util.function.ToLongFunction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -200,20 +202,57 @@ public Flux onReasoning( // next creates a new model call; the turn, // execution and writer stay unchanged. return next.apply( - new ReasoningInput( - candidate - .input() - .messages(), - candidate.input().tools(), - candidate - .input() - .options())); + new ReasoningInput( + candidate + .input() + .messages(), + candidate + .input() + .tools(), + candidate + .input() + .options())) + .onErrorResume( + retryError -> + Flux.error( + associateFirstOverflow( + error, + retryError))); }); }); } }; } + /** + * Keeps the first provider overflow associated with a failed single retry without replacing the + * retry's own error: as the primary cause when the cause slot is free, otherwise as a + * suppressed exception. Cancellations, fatal errors and interruptions are never rewritten so + * Reactor cancellation semantics survive. + */ + private static Throwable associateFirstOverflow(Throwable first, Throwable retryError) { + if (retryError == first + || retryError.getCause() == first + || retryError instanceof CancellationException + || Exceptions.isCancel(retryError) + || Exceptions.isFatal(retryError) + || ExceptionUtils.containsInterruptedException(retryError)) return retryError; + for (Throwable suppressed : retryError.getSuppressed()) { + if (suppressed == first) return retryError; + } + if (retryError.getCause() == null) { + try { + retryError.initCause(first); + return retryError; + } catch (IllegalStateException alreadyInitialized) { + // Providers build errors with an explicit null cause; that slot rejects + // initCause(), so fall through and attach the overflow as suppressed. + } + } + retryError.addSuppressed(first); + return retryError; + } + public static boolean isContextOverflowError(Throwable error) { String message = error.getMessage(); if (message == null) return false; diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/CompactionMiddlewareTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/CompactionMiddlewareTest.java index e18723c214..57ffb0eea1 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/CompactionMiddlewareTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/CompactionMiddlewareTest.java @@ -17,6 +17,9 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -39,9 +42,11 @@ import io.agentscope.core.util.JsonUtils; import io.agentscope.harness.agent.memory.compaction.CompactionConfig; import java.time.Duration; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CancellationException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -50,6 +55,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import reactor.core.publisher.Flux; +import reactor.core.publisher.Signal; import reactor.core.scheduler.Schedulers; import reactor.test.StepVerifier; @@ -245,6 +251,209 @@ protected Flux doStream( assertTrue(facts.stream().filter(e -> e.type().equals("model/end")).count() >= 2); } + /** + * A second provider failure must not erase the first overflow: the retry error stays terminal + * while the first overflow is attached as its primary cause, with no extra retry. + */ + @Test + void secondOverflowKeepsTheFirstOverflowAsPrimaryCause() { + AtomicInteger calls = new AtomicInteger(); + List overflows = new ArrayList<>(); + ChatModelBase model = + new ChatModelBase() { + public String getModelName() { + return "overflow-capture"; + } + + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + calls.incrementAndGet(); + var error = new IllegalStateException("context_length_exceeded"); + overflows.add(error); + return Flux.error(error); + } + }; + var recovery = + new CompactionMiddleware(null, new SuccessfulSummaryModel(), noCompactionConfig()); + var agent = + ReActAgent.builder() + .name("overflow-cause-test") + .model(model) + .sessionLogStore(new InMemorySessionLogStore()) + .middleware(recovery.overflowRecovery()) + .build(); + List> signals = + agent.streamEvents( + List.of(userMessage("old context"), userMessage("latest request")), + context("user", "overflow-cause-session")) + .materialize() + .collectList() + .block(Duration.ofSeconds(10)); + assertNotNull(signals); + Signal terminal = signals.get(signals.size() - 1); + assertTrue(terminal.isOnError(), "the retry failure must terminate the stream"); + assertEquals(2, overflows.size(), "exactly one retry is attempted"); + assertSame(overflows.get(1), terminal.getThrowable(), "the retry error stays terminal"); + assertSame( + overflows.get(0), + terminal.getThrowable().getCause(), + "the first overflow must remain the primary cause"); + } + + /** + * Provider errors built with an explicit null cause reject {@code initCause()}: the first + * overflow must fall back to suppressed and the retry error must stay terminal instead of being + * replaced by the resulting IllegalStateException. + */ + @Test + void nullCauseRetryErrorKeepsFirstOverflowAsSuppressed() { + AtomicInteger calls = new AtomicInteger(); + List errors = new ArrayList<>(); + ChatModelBase model = + new ChatModelBase() { + public String getModelName() { + return "overflow-null-cause"; + } + + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + Throwable error = + calls.getAndIncrement() == 0 + ? new IllegalStateException("context_length_exceeded") + : new IllegalStateException("retry failed", null); + errors.add(error); + return Flux.error(error); + } + }; + var recovery = + new CompactionMiddleware(null, new SuccessfulSummaryModel(), noCompactionConfig()); + var agent = + ReActAgent.builder() + .name("overflow-null-cause-test") + .model(model) + .sessionLogStore(new InMemorySessionLogStore()) + .middleware(recovery.overflowRecovery()) + .build(); + List> signals = + agent.streamEvents( + List.of(userMessage("old context"), userMessage("latest request")), + context("user", "overflow-null-cause-session")) + .materialize() + .collectList() + .block(Duration.ofSeconds(10)); + assertNotNull(signals); + Signal terminal = signals.get(signals.size() - 1); + assertTrue(terminal.isOnError(), "the retry failure must terminate the stream"); + assertEquals(2, errors.size(), "exactly one retry is attempted"); + assertSame(errors.get(1), terminal.getThrowable(), "the retry error stays terminal"); + assertNull( + terminal.getThrowable().getCause(), + "the pre-initialized null cause cannot be replaced"); + assertEquals(1, terminal.getThrowable().getSuppressed().length); + assertSame( + errors.get(0), + terminal.getThrowable().getSuppressed()[0], + "the first overflow must be attached as suppressed"); + } + + /** A cancelled retry must keep Reactor cancellation semantics and never be rewritten. */ + @Test + void cancelledRetryKeepsCancellationUntouched() { + AtomicInteger calls = new AtomicInteger(); + List errors = new ArrayList<>(); + ChatModelBase model = + new ChatModelBase() { + public String getModelName() { + return "overflow-then-cancel"; + } + + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + Throwable error = + calls.getAndIncrement() == 0 + ? new IllegalStateException("context_length_exceeded") + : new CancellationException("user cancelled"); + errors.add(error); + return Flux.error(error); + } + }; + var recovery = + new CompactionMiddleware(null, new SuccessfulSummaryModel(), noCompactionConfig()); + var agent = + ReActAgent.builder() + .name("overflow-cancel-test") + .model(model) + .sessionLogStore(new InMemorySessionLogStore()) + .middleware(recovery.overflowRecovery()) + .build(); + List> signals = + agent.streamEvents( + List.of(userMessage("old context"), userMessage("latest request")), + context("user", "overflow-cancel-session")) + .materialize() + .collectList() + .block(Duration.ofSeconds(10)); + assertNotNull(signals); + Signal terminal = signals.get(signals.size() - 1); + assertTrue(terminal.isOnError(), "the cancellation must surface as the terminal signal"); + assertEquals(2, errors.size()); + assertSame(errors.get(1), terminal.getThrowable()); + assertNull( + terminal.getThrowable().getCause(), + "a cancelled retry must not be rewritten with the first overflow"); + } + + /** + * When the emergency summarization itself fails, the first overflow must still be the + * reported error and no provider retry may run. + */ + @Test + void failedEmergencySummaryStillReportsTheFirstOverflow() { + AtomicInteger calls = new AtomicInteger(); + List errors = new ArrayList<>(); + ChatModelBase model = + new ChatModelBase() { + public String getModelName() { + return "overflow-no-retry"; + } + + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + calls.incrementAndGet(); + Throwable error = new IllegalStateException("context_length_exceeded"); + errors.add(error); + return Flux.error(error); + } + }; + var recovery = + new CompactionMiddleware( + null, + new FailingSummaryModel(new IllegalStateException("summary unavailable")), + noCompactionConfig()); + var agent = + ReActAgent.builder() + .name("overflow-summary-failure-test") + .model(model) + .sessionLogStore(new InMemorySessionLogStore()) + .middleware(recovery.overflowRecovery()) + .build(); + List> signals = + agent.streamEvents( + List.of(userMessage("old context"), userMessage("latest request")), + context("user", "overflow-summary-failure-session")) + .materialize() + .collectList() + .block(Duration.ofSeconds(10)); + assertNotNull(signals); + Signal terminal = signals.get(signals.size() - 1); + assertTrue(terminal.isOnError()); + assertSame( + errors.get(0), + terminal.getThrowable(), + "the first overflow stays the reported error when summarization fails"); + assertEquals(1, calls.get(), "no provider retry may run after a failed emergency summary"); + } + /** An ordinary compaction failure skips compaction and invokes original reasoning once. */ @Test void ordinaryCompactionFailureFallsBackToOriginalInputOnce() { diff --git a/docs/v2/en/docs/harness/session-log.md b/docs/v2/en/docs/harness/session-log.md index d12e55a503..4ac8de8f93 100644 --- a/docs/v2/en/docs/harness/session-log.md +++ b/docs/v2/en/docs/harness/session-log.md @@ -260,6 +260,8 @@ The catalog below helps locate the facts needed for a timeline, diagnosis or app Events cover adapter-visible requests and output, not provider internals or backups of referenced files. Keep raw requests, tool results and checkpoints in an authorized diagnostic interface; filter what you expose to a browser. +The `request/prepared` payload records the model input, but media blocks (image, audio, video and data, including media nested in tool results) are replaced by a `redacted` placeholder. Binary data, URLs and file paths never enter the log, while the model still receives the real payload. + The `compaction/end` payload includes `compactionId`, `status`, `beforeMsgCount`, and `beforeTokenCount`. When `status` is `completed`, it also includes `afterMsgCount` and `afterTokenCount`, calculated from the effective post-compaction conversation. Failed or cancelled diff --git a/docs/v2/zh/docs/harness/session-log.md b/docs/v2/zh/docs/harness/session-log.md index bac4fc0664..469b6fed90 100644 --- a/docs/v2/zh/docs/harness/session-log.md +++ b/docs/v2/zh/docs/harness/session-log.md @@ -294,6 +294,8 @@ for (var event : log.scan(0, through)) { 记录范围是模型适配器可见的请求与输出、工具和框架执行过程，不包括模型服务内部过程，也不会自动备份被引用的文件。原始请求和工具结果可能包含业务数据，面向浏览器的接口应挑选需要展示的字段。 +`request/prepared` 记录模型输入，但其中的媒体块（图片、音频、视频、数据块，包括工具结果中嵌套的媒体）会替换为 `redacted` 占位符。二进制数据、URL 和文件路径不会进入日志，模型仍然收到真实内容。 + `compaction/end` 的载荷包含 `compactionId`、`status`、`beforeMsgCount` 和 `beforeTokenCount`。当 `status` 为 `completed` 时，还会包含根据压缩后有效对话计算的 `afterMsgCount` 和 `afterTokenCount`。压缩失败或取消时不会伪造压缩后指标，因为此时没有可用的压缩后上下文。