This implementation implements a Reasoning and Acting Agent as a Temporal Workflow that supports potentially infinite runtime thanks to the use of continue-as-new for context compaction and long-term memory managment and retrieval activities. The agent processes incoming messages, maintains a context of the conversation, and decides when to comp…

ActivitiesImpl.java

public class ActivitiesImpl implements Activities {
	@Override
	public RetrieveMemoryRecordsResult retrieveMemoryRecordsActivity(String query,
			List<MemoryStrategyType> strategyTypes)
			throws ApplicationFailure {
		try {
			RetrieveMemoryRecordsResponse response = AgentCoreMemory.retrieveMemoryRecords(query, strategyTypes);
			if (!response.hasMemoryRecordSummaries()) {
				return new RetrieveMemoryRecordsResult(List.of()); // empty
			}

			var memoryRecords = new ArrayList<LabeledMemoryRecord>();

			for (MemoryRecordSummary memoryRecordSummary : response.memoryRecordSummaries()) {
				MemoryStrategyType memoryStrategyType = AgentCoreMemory.getMemoryStrategyType(memoryRecordSummary);
				MemoryContent memoryContent = memoryRecordSummary.content();
				if (memoryContent.type() != MemoryContent.Type.TEXT) {
					continue;
				}

				String text = memoryContent.text();
				memoryRecords.add(new LabeledMemoryRecord(memoryStrategyType, text));
			}
			return new RetrieveMemoryRecordsResult(memoryRecords);
		} catch (Exception e) {
			System.err.println("Error in AgentCoreMemory.retrieveMemoryRecords: " + e.getMessage());
			throw ApplicationFailure.newFailure("AgentCoreMemory.retrieveMemoryRecords failed: " + e.getMessage(),
					"RetrieveAgentCoreMemoryRecordsActivityError");
		}
	}

	@Override
	public void persistMemoryActivity(List<ContextEntry> entries, String sessionId) throws ApplicationFailure {
		// Send the messages to the AgentCore memory extraction system.
		try {
			AgentCoreMemory.createEvent(entries);
		} catch (Exception e) {
			System.err.println("Error in AgentCoreMemory.createEvent: " + e.getMessage());
			throw ApplicationFailure.newFailure("AgentCoreMemory.createEvent failed: " + e.getMessage(),
					"PersistAgentCoreMemoryActivityError");
		}
	}
}

AgentMemoryWorkflowImpl.java

package bitovi.workflow;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

import org.json.JSONObject;

import bitovi.activities.Activities;
import bitovi.activities.types.ActionDetail;
import bitovi.activities.types.CompactResponse;
import bitovi.activities.types.LabeledMemoryRecord;
import bitovi.activities.types.ObservationResponse;
import bitovi.activities.types.RetrieveMemoryRecordsResult;
import bitovi.activities.types.ThoughtResponse;
import bitovi.workflow.types.ContextEntry;
import bitovi.workflow.types.ContextEntryType;
import bitovi.workflow.types.ContinueAsNewState;
import bitovi.workflow.types.MessagePayload;
import bitovi.workflow.types.UsageMetadata;
import bitovi.workflow.types.WorkflowInput;
import bitovi.workflow.types.WorkflowResult;
import io.temporal.activity.ActivityOptions;
import io.temporal.common.RetryOptions;
import io.temporal.workflow.Workflow;
import software.amazon.awssdk.services.bedrockagentcorecontrol.model.MemoryStrategyType;

public class AgentMemoryWorkflowImpl implements AgentMemoryWorkflow {
	private final ActivityOptions defaultActivityOptions = ActivityOptions
			.newBuilder()
			.setStartToCloseTimeout(Duration.ofMinutes(5))
			.setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(3).build())
			.build();

	private final Activities activities = Workflow.newActivityStub(Activities.class, defaultActivityOptions);

	private static final int COMPACTION_CONTEXT_TOKEN_THRESHOLD = 102400;

	// Signal state
	private final List<MessagePayload> pending = new ArrayList<>();
	private boolean userRequestedExit = false;
	private boolean userRequestedContinueAsNew = false;

	@Override
	public void receiveMessage(MessagePayload payload) {
		pending.add(payload);
	}

	@Override
	public void requestExit() {
		userRequestedExit = true;
	}

	@Override
	public void requestContinueAsNew() {
		userRequestedContinueAsNew = true;
	}

	@Override
	public WorkflowResult execute(WorkflowInput input) {
		// Initialize state from input
		List<ContextEntry> context = input.continueAsNew() != null ? input.continueAsNew().context()
				: new ArrayList<>();
		List<UsageMetadata> usage = input.continueAsNew() != null ? input.continueAsNew().usage() : new ArrayList<>();

		// If continuing as new, restore pending messages
		if (input.continueAsNew() != null && input.continueAsNew().pending() != null) {
			pending.addAll(input.continueAsNew().pending());
		}

		// Wait for first message, exit, or compaction request
		Workflow.await(() -> !pending.isEmpty() || userRequestedExit || userRequestedContinueAsNew);

		// Main event loop
		while (true) {
			// Check if exit requested
			if (userRequestedExit) {
				// Aggregate usage metrics and return
				UsageMetadata finalUsage = usage.stream()
						.reduce(new UsageMetadata(0, 0, 0),
								(acc, curr) -> new UsageMetadata(
										acc.inputTokens() + curr.inputTokens(),
										acc.outputTokens() + curr.outputTokens(),
										acc.totalTokens() + curr.totalTokens()));
				return new WorkflowResult(finalUsage);
			}

			// Check if the user or temporal recommended continueAsNew
			boolean shouldContinueAsNew = (userRequestedContinueAsNew || Workflow.getInfo().isContinueAsNewSuggested());

			// Check if context token usage exceeds threshold
			Integer contextLength = activities.getTokenUsage(context);
			Workflow.getLogger(AgentMemoryWorkflowImpl.class).info("Current context token usage: " + contextLength);

			if (shouldContinueAsNew || (contextLength != null && contextLength > COMPACTION_CONTEXT_TOKEN_THRESHOLD)) {
				CompactResponse compactResponse = activities.compactActivity(context);

				// Track usage
				if (compactResponse.usage() != null) {
					usage.add(compactResponse.usage());
				}

				// Create continue-as-new state
				ContinueAsNewState continueAsNewState = new ContinueAsNewState(
						compactResponse.context(),
						usage,
						new ArrayList<>(pending));

				// Continue as new
				Workflow.continueAsNew(new WorkflowInput(continueAsNewState));
			}

			// Process all pending messages
			if (!pending.isEmpty()) {

				for (MessagePayload msg : pending) {
					// Add user message to context as structured entry
					ContextEntry userEntry = ContextEntry.fromUser(msg.message());
					context.add(userEntry);
				}

				// Clear pending messages
				pending.clear();
			}

			// Start ReAct (Reasoning and Acting) Steps
			String query = context.stream()
					.map(ContextEntry::toXMLString)
					.collect(Collectors.joining("\n"));

			List<MemoryStrategyType> memoryStrategies = List.of(
					MemoryStrategyType.USER_PREFERENCE,
					MemoryStrategyType.SEMANTIC,
					MemoryStrategyType.SUMMARIZATION);

			// Retrieve Long Term Memories
			RetrieveMemoryRecordsResult retrieveResult = activities.retrieveMemoryRecordsActivity(query,
					memoryStrategies);
			List<LabeledMemoryRecord> memoryRecords = retrieveResult.memories();

			// Get thought from AI based on context and retrieved memories
			ThoughtResponse thoughtResponse = activities.thoughtActivity(context, memoryRecords);

			// Track usage
			if (thoughtResponse.usage() != null) {
				usage.add(thoughtResponse.usage());
			}

			// Check response type
			if (thoughtResponse.type().equals("answer")) {
				// Answer type - add to context and wait for next message
				ContextEntry answerEntry = ContextEntry.fromAnswer(thoughtResponse.answer());

				context.add(answerEntry);

				// Collect entries from most recent USER_MESSAGE to ANSWER
				List<ContextEntry> persist = new ArrayList<>();
				context.forEach((action) -> {
					if (action.type().equals(ContextEntryType.USER_MESSAGE)
							|| action.type().equals(ContextEntryType.ANSWER)) {
						persist.add(action);
					}
				});

				activities.persistMemoryActivity(persist, Workflow.getInfo().getRunId());

				// Once the agent has generated an answer, we wait for the next user message or
				// other signal request before continuing
				Workflow.await(() -> !pending.isEmpty() || userRequestedExit || userRequestedContinueAsNew);
			}

			if (thoughtResponse.type().equals("action")) {
				// Action type - execute action and get observation
				ActionDetail action = thoughtResponse.action();

				// Add thought to context
				ContextEntry thoughtEntry = ContextEntry.fromThought(thoughtResponse.thought());
				context.add(thoughtEntry);

				// Serialize action input for context
				String actionInputJson;
				try {
					actionInputJson = new JSONObject(action.input().parameters()).toString();
				} catch (Exception e) {
					actionInputJson = "{}";
				}

				// Add action to context
				ContextEntry actionEntry = ContextEntry.fromAction(action.reason(), action.name(), actionInputJson);
				context.add(actionEntry);

				// Execute the action
				String actionResult = activities.actionActivity(action.name(), action.input());

				// Get observation
				ObservationResponse observationResponse = activities.observationActivity(thoughtResponse.thought(),
						action.name(), actionInputJson, actionResult);

				// Track usage
				if (observationResponse.usage() != null) {
					usage.add(observationResponse.usage());
				}

				// Add observation to context
				ContextEntry observationEntry = ContextEntry.fromObservation(observationResponse.observations());
				context.add(observationEntry);
			}
		}
	}
}

ThoughtActivity.java

public class ThoughtActivity {
	public static ThoughtResponse execute(String promptTemplate, List<ContextEntry> context,
			List<LabeledMemoryRecord> memoryRecords)
			throws ApplicationFailure {
		try {
			// Convert ContextEntry list to XML strings for LLM prompt
			List<String> contextStrings = context.stream()
					.map(ContextEntry::toXMLString)
					.collect(Collectors.toList());

			// Get current date
			String currentDate = LocalDate.now().toString();

			// Get available tools as XML string
			String availableActions = ToolRegistry.getToolsAsXmlString();

			// Convert LabeledMemoryRecord list to XML strings for LLM prompt
			List<String> memoryRecordStrings = memoryRecords.stream()
					.map(LabeledMemoryRecord::toXMLString)
					.collect(Collectors.toList());

			// Format prompt with placeholders
			String systemPrompt = promptTemplate
					.replace("{currentDate}", currentDate)
					.replace("{previousSteps}", String.join("\n", contextStrings))
					.replace("{memoryRecords}", String.join("\n", memoryRecordStrings))
					.replace("{availableActions}", availableActions);

			// Call Bedrock with high-quality model
			Config config = new Config();
			String modelId = config.getProperty("AWS_MODEL_ID");

			ModelResponseWithUsage response = BedrockConverse.bedrockConverseWithUsage(
					systemPrompt,
					null, // No tool config needed for thought
					modelId);

			String responseText = response.response();
			if (responseText == null || responseText.isEmpty()) {
				throw ApplicationFailure.newFailure("Empty response from model", "EmptyModelResponse");
			}

			return ThoughtResponse.from(response);

		} catch (JSONException e) {
			String errorMsg = "Error parsing JSON response: " + e.getMessage();
			System.err.println(errorMsg);
			EventClient.emitEvent("error", "Thought error: " + errorMsg);
			throw ApplicationFailure.newFailure("Failed to parse model response: " + e.getMessage(),
					"ThoughtActivityError");
		} catch (ApplicationFailure e) {
			String errorMsg = "Error in thoughtActivity: " + e.getMessage();
			System.err.println(errorMsg);
			EventClient.emitEvent("error", "Thought error: " + errorMsg);
			throw ApplicationFailure.newFailure("thoughtActivity failed: " + e.getMessage(),
					"ThoughtActivityError");
		}
	}
}