feat: add should_complete to CompletionConfig - #605
Conversation
862e017 to
c72785b
Compare
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
c72785b to
306c81c
Compare
306c81c to
5483a62
Compare
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
5483a62 to
fffee6d
Compare
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
fffee6d to
7ad4242
Compare
7ad4242 to
302330d
Compare
05ea8fa to
84b7cb7
Compare
This comment has been minimized.
This comment has been minimized.
84b7cb7 to
de4ef87
Compare
This comment has been minimized.
This comment has been minimized.
de4ef87 to
28943d5
Compare
28943d5 to
2480b1e
Compare
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
| succeeded, | ||
| failed, | ||
| items_snapshot, | ||
| restored_events_seen >= len(restored_terminal_indexes), |
There was a problem hiding this comment.
Codex AI review
[P1] Block new work while restoring terminal events. The restoration gate only suppresses predicate evaluation; scheduling remains enabled. With two restored terminal branches and max_concurrency=2, the first replayed event frees a slot and causes a previously unstarted third branch to execute before the second event opens the gate. This can introduce new durable side effects during replay. Track previously existing branches and admit only those until restoration and predicate evaluation are complete.
| # Defer predicate evaluation until all previously-checkpointed | ||
| # terminal branches have reported again, ensuring deterministic | ||
| # decisions regardless of replayed event order. | ||
| restored_terminal_indexes: set[int] = set() | ||
| if self.policy.should_complete is not None: | ||
| for exe in self.executables: | ||
| op_id: str = self.operation_id_namespace.create_id_for_step( | ||
| exe.index | ||
| ) | ||
| cp: CheckpointedResult = execution_state.get_checkpoint_result( | ||
| op_id | ||
| ) | ||
| if cp.is_succeeded() or cp.is_failed(): | ||
| restored_terminal_indexes.add(exe.index) | ||
| restored_events_seen: int = 0 |
There was a problem hiding this comment.
Codex AI review
[P1] Persist the custom completion decision itself. Re-evaluating only after all terminal child checkpoints replay does not reproduce the status that originally triggered the decision. For example, a deterministic predicate that completes when success_count == 1 can fire live while a second branch has already checkpointed; if the invocation dies before the parent checkpoint, resume evaluates at two successes and continues with a different result. Synchronously checkpoint the decision/outcome when it fires, or explicitly restrict and validate predicates to monotonic decisions with stable outcomes.
Codex AI reviewTwo replay-safety issues can change execution after a crash. Existing resume tests do not cover the decision-to-parent-checkpoint race window. Reviewed commit |
Issue #, if available: #519
Description of changes:
Adds a
should_completepredicate to CompletionConfig for map and parallel operations, giving full control over when a batch completes early.The predicate receives a CompletionStatus snapshot (counts + per-item statuses) and returns a CompletionDecision - either continue_batch() or complete_batch(outcome). The outcome determines whether
CUSTOM_COMPLETION_SUCCEEDEDorCUSTOM_COMPLETION_FAILEDis reported.Key design decisions -
Testing -
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.