Skip to content

fix(harness): unify compaction overflow recovery and redact session log media sources #3568

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dargoner wants to merge 2 commits into
base: main
Choose a base branch
from
+484 −9
Open

fix(harness): unify compaction overflow recovery and redact session log media sources #3568

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
436bd3b
fix(harness): redact media in prepared-request logs and chain the fir…
dargoner Oct 29, 2026
e78b44f
Merge remote-tracking branch 'origin/main' into fix/compaction-recove…
dargoner Oct 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 100 additions & 1 deletion agentscope-core/src/main/java/io/agentscope/core/session/SessionRecorder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -599,7 +606,7 @@ public Mono<Void> 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);
Expand All @@ -610,4 +617,96 @@ public Mono<Void> 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.
*
* <p>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<Msg> redactedForLog(List<Msg> messages) {
if (messages == null) return List.of();
List<Msg> redacted = new ArrayList<>(messages.size());
for (Msg message : messages) redacted.add(redactMessage(message));
return redacted;
}

private static Msg redactMessage(Msg message) {
List<ContentBlock> content = message.getContent();
if (content == null || content.stream().noneMatch(SessionRecorder::containsMedia))
return message;
List<ContentBlock> 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<ContentBlock> 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<ContentBlock> output = result.getOutput();
if (output == null || output.stream().noneMatch(SessionRecorder::containsMedia))
return result;
List<ContentBlock> 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);
}
}
118 changes: 118 additions & 0 deletions agentscope-core/src/test/java/io/agentscope/core/session/ReActSessionLogTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<ChatResponse> stream(
List<Msg> messages, List<ToolSchema> 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();
Expand Down
55 changes: 47 additions & 8 deletions ...pe-harness/src/main/java/io/agentscope/harness/agent/middleware/CompactionMiddleware.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -200,20 +202,57 @@ public Flux<AgentEvent> 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;
Expand Down
Loading
Loading