Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
52 changes: 46 additions & 6 deletions a2a/src/main/java/com/google/adk/a2a/executor/AgentExecutor.java
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,15 @@ public class AgentExecutor implements io.a2a.server.agentexecution.AgentExecutor
private static final Logger logger = LoggerFactory.getLogger(AgentExecutor.class);
private static final String USER_ID_PREFIX = "A2A_USER_";
private static final String A2A_METADATA_KEY = "a2a_metadata";

/**
* Env var that puts exception text back into the failure message sent to the peer.
*
* <p>Off by default, and meant for local debugging only: enabling it on a network-reachable
* deployment restores the disclosure {@link #failedMessage} exists to prevent.
*/
private static final String DEBUG_ERRORS_ENV_VAR = "ADK_DEBUG_ERRORS";

private final Map<String, Disposable> activeTasks = new ConcurrentHashMap<>();
private final Runner.Builder runnerBuilder;
private final AgentExecutorConfig agentExecutorConfig;
Expand Down Expand Up @@ -233,11 +242,8 @@ public void execute(RequestContext ctx, EventQueue eventQueue) {
.materialize()
.flatMapCompletable(
notification -> {
Throwable error = notification.getError();
if (error != null) {
logger.error("Runner failed to execute", error);
}
return handleExecutionEnd(ctx, error, eventQueue);
// The failure is logged, with its correlation id, by failedMessage().
return handleExecutionEnd(ctx, notification.getError(), eventQueue);
})
.doFinally(() -> cleanupTask(ctx.getTaskId()))
.subscribe(
Expand Down Expand Up @@ -304,16 +310,50 @@ private Maybe<Session> prepareSession(
}));
}

/**
* Builds the failure message handed back to the remote peer.
*
* <p>The peer is not trusted with {@code e.getMessage()}: exception text routinely names absolute
* filesystem paths, class and module locations, configuration values and echoed request payloads,
* none of which the caller needs and all of which are useful reconnaissance. The throwable is
* logged here in full next to a short opaque id, and the peer gets a fixed summary plus that same
* id, so an operator handed the id can find the real stack trace. Set {@code ADK_DEBUG_ERRORS=1}
* to put the exception text back into the response while debugging locally.
*/
private static Message failedMessage(RequestContext context, Throwable e) {
String errorId = UUID.randomUUID().toString();
logger.error("Runner failed to execute [errorId={}]", errorId, e);
return new Message.Builder()
.messageId(UUID.randomUUID().toString())
.contextId(context.getContextId())
.taskId(context.getTaskId())
.role(Message.Role.AGENT)
.parts(ImmutableList.of(new TextPart(e.getMessage())))
.parts(ImmutableList.of(new TextPart(failureText(e, errorId, debugErrorsEnabled()))))
.build();
}

/**
* Returns the failure text that is safe to hand to the remote peer.
*
* @param includeDetail whether to append the exception type and message; see {@link
* #DEBUG_ERRORS_ENV_VAR}.
*/
static String failureText(Throwable e, String errorId, boolean includeDetail) {
String text = "Agent execution failed. (error_id: " + errorId + ")";
if (includeDetail) {
text = text + ": " + e.getClass().getName() + ": " + e.getMessage();
}
return text;
}

private static boolean debugErrorsEnabled() {
String value = System.getenv(DEBUG_ERRORS_ENV_VAR);
if (value == null) {
return false;
}
return value.equals("1") || value.equalsIgnoreCase("true");
}

// Processor that will process all events related to the one runner invocation.
private static class EventProcessor {
private final String runArtifactId;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,10 @@
@RunWith(JUnit4.class)
public final class AgentExecutorTest {

/** A throwable message shaped like the ones that leak host detail. */
private static final String SECRET_ERROR =
"Runner error: /home/victim/.config/adk/credentials.json (No such file)";

private EventQueue eventQueue;
private List<Object> enqueuedEvents;
private TestAgent testAgent;
Expand Down Expand Up @@ -137,7 +141,7 @@ public void createAgentExecutor_noAgentExecutorConfig_throwsException() {
public void execute_withBeforeExecuteCallback_cancelsExecutionOnError() {
// If callback returns error, execution should stop/fail.
Callbacks.BeforeExecuteCallback callback =
ctx -> Single.error(new RuntimeException("Cancelled"));
ctx -> Single.error(new RuntimeException(SECRET_ERROR));

AgentExecutorConfig config =
AgentExecutorConfig.builder().beforeExecuteCallback(callback).build();
Expand All @@ -162,7 +166,10 @@ public void execute_withBeforeExecuteCallback_cancelsExecutionOnError() {
assertThat(statusEvent.getStatus().state().toString()).isEqualTo("FAILED");
assertThat(statusEvent.getStatus().message().getParts().get(0)).isInstanceOf(TextPart.class);
TextPart textPart = (TextPart) statusEvent.getStatus().message().getParts().get(0);
assertThat(textPart.getText()).contains("Cancelled");
// The remote peer gets a correlation id for the logged throwable, not its
// message -- see AgentExecutor#failedMessage.
assertThat(textPart.getText()).startsWith("Agent execution failed. (error_id: ");
assertThat(textPart.getText()).doesNotContain(SECRET_ERROR);
}

@Test
Expand Down Expand Up @@ -289,7 +296,7 @@ public void execute_withAfterExecuteCallback_modifiesStatus() {

@Test
public void execute_runnerFails_registersFailedEvent() {
testAgent.setEventsToEmit(Flowable.error(new RuntimeException("Runner error")));
testAgent.setEventsToEmit(Flowable.error(new RuntimeException(SECRET_ERROR)));
AgentExecutor executor =
new AgentExecutor.Builder()
.agentExecutorConfig(AgentExecutorConfig.builder().build())
Expand All @@ -316,7 +323,28 @@ public void execute_runnerFails_registersFailedEvent() {
assertThat(statusEvent.getStatus().state()).isEqualTo(TaskState.FAILED);
assertThat(statusEvent.getStatus().message().getParts().get(0)).isInstanceOf(TextPart.class);
TextPart textPart = (TextPart) statusEvent.getStatus().message().getParts().get(0);
assertThat(textPart.getText()).isEqualTo("Runner error");
// A runner failure is reported to the peer as a correlation id only: the
// throwable's message names host paths and is for the server log.
assertThat(textPart.getText()).startsWith("Agent execution failed. (error_id: ");
assertThat(textPart.getText()).doesNotContain(SECRET_ERROR);
assertThat(textPart.getText()).doesNotContain("/home/victim");
}

@Test
public void failureText_withoutDebug_carriesOnlyTheCorrelationId() {
String text = AgentExecutor.failureText(new RuntimeException(SECRET_ERROR), "abc-123", false);

assertThat(text).isEqualTo("Agent execution failed. (error_id: abc-123)");
}

@Test
public void failureText_withDebug_carriesTheThrowableDetail() {
// ADK_DEBUG_ERRORS=1 is the documented opt-in for local debugging.
String text = AgentExecutor.failureText(new RuntimeException(SECRET_ERROR), "abc-123", true);

assertThat(text).startsWith("Agent execution failed. (error_id: abc-123): ");
assertThat(text).contains("java.lang.RuntimeException");
assertThat(text).contains(SECRET_ERROR);
}

@Test
Expand Down
Loading