xintongsong commented on code in PR #422:
URL: https://github.com/apache/flink-agents/pull/422#discussion_r2685076267
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/context/PythonRunnerContextImpl.java:
##########
@@ -48,4 +78,120 @@ public void sendEvent(String type, byte[] event, String
eventString) {
// this method will be invoked by PythonActionExecutor's python
interpreter.
sendEvent(new PythonEvent(event, type, eventString));
}
+
+ /**
+ * Initializes the recovery state for the current action.
+ *
+ * @param actionState the ActionState for the current action (used for
adding CallRecords)
+ * @param actionStatePersister callback to persist ActionState after each
code block completion
+ */
+ public void initRecoveryState(ActionState actionState, Runnable
actionStatePersister) {
+ this.currentCallIndex = 0;
+ this.currentActionState = actionState;
+ this.actionStatePersister = actionStatePersister;
+ this.recoveryCallRecords =
+ actionState != null && actionState.getCallRecords() != null
+ ? new ArrayList<>(actionState.getCallRecords())
+ : new ArrayList<>();
+ }
+
+ /**
+ * Gets the current call index.
+ *
+ * @return the current call index
+ */
+ public int getCurrentCallIndex() {
+ return currentCallIndex;
+ }
+
+ /**
+ * Tries to get a cached CallRecord for recovery.
+ *
+ * <p>This method is called by Python's execute/execute_async to check if
a previous result
+ * exists. If found and validated, the cached result is returned.
+ *
+ * @param functionId the function identifier
+ * @param argsDigest the digest of serialized arguments
+ * @return array containing [isHit (boolean), resultPayload (byte[]),
exceptionPayload
+ * (byte[])], or null if miss
+ */
+ public Object[] tryGetCachedCallRecord(String functionId, String
argsDigest) {
+ mailboxThreadChecker.run();
+
+ if (currentCallIndex < recoveryCallRecords.size()) {
+ CallRecord record = recoveryCallRecords.get(currentCallIndex);
+
+ if (record.matches(functionId, argsDigest)) {
+ LOG.debug(
+ "CallRecord hit at index {}: functionId={},
argsDigest={}",
+ currentCallIndex,
+ functionId,
+ argsDigest);
+ currentCallIndex++;
+ return new Object[] {true, record.getResultPayload(),
record.getExceptionPayload()};
+ } else {
+ // Non-deterministic call detected, clear subsequent records
+ LOG.warn(
Review Comment:
@yanand0909 , is there any concrete use case you see that makes failing fast
necessary? I'm asking because I just can't think of why a user may want to fail
the agent simply because it behaves differently before / after the recovery.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
