Repository navigation
Expand file tree
/
Copy pathclass-wp-agent-workflow-scoped-drain.php
More file actions
800 lines (735 loc) · 29.3 KB
/
Copy pathclass-wp-agent-workflow-scoped-drain.php
File metadata and controls
800 lines (735 loc) · 29.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
<?php
/**
* A generic, scoped Action Scheduler drain — the foreground pump that completes a
* fan-out on ANY runtime, whether or not WP-Cron ever fires.
*
* ## Why this exists
*
* The Action Scheduler branch executor ({@see WP_Agent_Workflow_Action_Scheduler_Branch_Executor})
* dispatches each parallel branch as a claimed AS async action and then relies on
* something EXTERNAL to pump the AS queue — WP-Cron, or a runtime cron heartbeat —
* to actually claim and run those actions. That is fine on a runtime where WP-Cron
* fires. It is NOT fine on a managed host that sets `DISABLE_WP_CRON` (e.g. WP
* Cloud) or a core/CLI context with no cron heartbeat at all: the branches enqueue,
* nothing claims them, and the run strands SUSPENDED forever even though the
* branches are perfectly runnable. A manual `wp action-scheduler run` drains them —
* proving the branches CAN run; they just need something to actively pump the queue.
*
* This service IS that pump, in-process. It synchronously drains a SCOPED set of AS
* actions (by hook + group) until the scope is empty or a terminal condition is met,
* with wall-clock / batch / memory budgets so it cannot run away. A foreground
* orchestrator (a CLI, a `wp-cron.php` request, or an `await=true` foreground poll)
* calls it to make its own fan-out complete without depending on WP-Cron.
*
* ## A generic, product-free primitive
*
* This is the generic core of a scoped-drain mechanism: a synchronous, in-process
* pump over a caller-scoped set of Action Scheduler actions, with budget / memory /
* timeout discipline and a terminal-status callback. It carries NO product- or
* domain-specific coupling — no fixed group or hook constants, no job-id or
* worker-lane scoping, no per-action config lookups, no product-prefixed filter
* names. Scope is a caller-passed hook list + group (defaulting to the branch
* executor's own hooks + group), tuning is via `agents_`-prefixed filters, and the
* terminal condition is a caller-supplied callback. Every caller of the branch
* executor can share this one mechanism; a caller that needs richer scoping layers
* it on top rather than pushing it down here.
*
* ## CRITICAL — foreground / orchestrator context ONLY. Never from a branch worker.
*
* This MUST only be called from a FOREGROUND / orchestrator context:
*
* - a CLI command awaiting the run,
* - a `wp-cron.php` request,
* - the `await=true` foreground poll of a suspended run.
*
* It MUST NEVER be called from inside a branch worker draining its own siblings. A
* branch action runs while its worker HOLDS an Action Scheduler claim; if that
* worker then tried to drain the sibling branches, it would hold a claim while
* pumping the queue that the siblings themselves need claims from — re-introducing
* the exact self-deadlock that dispatch-and-return was built to avoid (the parent
* blocks on children it is itself starving). {@see drain()} DETECTS this: if it is
* invoked while WordPress is already firing one of the scoped hooks (i.e. this
* request is itself a claimed branch/resume action, observable via `doing_action()`),
* it refuses and returns a `refused_reentrant` result rather than deadlocking. A
* process-level re-entrancy guard additionally prevents a nested drain within one
* request.
*
* @package AgentsAPI
* @since 0.5.2
*/
namespace AgentsAPI\AI\Workflows;
defined( 'ABSPATH' ) || exit;
/**
* Drain a caller-scoped set of Action Scheduler actions in-process, with budgets.
*/
final class WP_Agent_Workflow_Scoped_Drain {
/**
* Process-level re-entrancy guard. A drain that is already running in this
* request must not be re-entered (e.g. a branch action that itself somehow
* reaches back into a drain). Paired with the `doing_action()` check in
* {@see self::guard_context()} this makes the self-deadlock unreachable.
*
* @since 0.5.2
* @var bool
*/
private static bool $draining = false;
/**
* Whether Action Scheduler's in-process runner API is available. This — not the
* mere presence of the plugin — is the gate: the drain needs the store, the
* runner, and the query API all present to claim and run actions in-process.
*
* @since 0.5.2
*/
public static function is_available(): bool {
return class_exists( '\ActionScheduler' )
&& class_exists( '\ActionScheduler_Store' )
&& function_exists( 'as_get_scheduled_actions' );
}
/**
* The default hook scope for a workflow fan-out drain: the executor's per-branch
* hook plus its resume hook, read from the executor constants (never hardcoded)
* so this class and the executor can never drift.
*
* @since 0.5.2
*
* @return array<int,string> Default scoped hooks.
*/
public static function default_hooks(): array {
return array(
WP_Agent_Workflow_Action_Scheduler_Branch_Executor::BRANCH_HOOK,
WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RECONCILE_HOOK,
WP_Agent_Workflow_Action_Scheduler_Branch_Executor::AGGREGATE_HOOK,
WP_Agent_Workflow_Action_Scheduler_Branch_Executor::RESUME_HOOK,
);
}
/**
* The default AS group for a workflow fan-out drain, read from the executor
* (which shares its group with the cron bridge) so scope stays in one place.
*
* @since 0.5.2
*/
public static function default_group(): string {
return WP_Agent_Workflow_Action_Scheduler_Branch_Executor::GROUP;
}
/**
* Synchronously drain the scoped Action Scheduler actions until the scope is
* empty, a budget is exhausted, or the terminal-status callback signals the run
* is done — whichever comes first.
*
* The loop, each iteration:
* 1. asks the terminal-status callback whether the run is already done (so a
* completed run stops immediately, not after wasting a batch);
* 2. checks the memory soft limit and the wall-clock budget;
* 3. claims and runs ONE batch of due scoped actions in-process via
* {@see self::run_batch()} — pure Action Scheduler mechanics
* (`\ActionScheduler_Store::instance()->stake_claim()` +
* `\ActionScheduler::runner()->process_action()`);
* 4. stops if a batch made no observable progress (guards against a stuck
* scope spinning the loop).
*
* @since 0.5.2
*
* @param array{
* hooks?:array<int,string>,
* group?:string,
* batch_size?:int,
* limit?:int,
* time_limit_ms?:int,
* time_limit?:int,
* stop_before_timeout_ms?:int,
* stop_before_timeout?:int,
* allow_group_fallback?:bool,
* execution_context?:string,
* terminal_status_callback?:callable
* } $options Drain options.
* @return array<string,int|string|bool|array{code:string,message:string,actionable:bool}> Drain stats.
*/
public function drain( array $options = array() ): array {
$hooks = $this->normalize_hooks( $options['hooks'] ?? null );
$group = isset( $options['group'] ) && '' !== (string) $options['group'] ? (string) $options['group'] : self::default_group();
$batch_size = max( 1, (int) ( $options['batch_size'] ?? 25 ) );
$limit = max( 0, (int) ( $options['limit'] ?? 0 ) );
$time_limit_ms = isset( $options['time_limit_ms'] )
? max( 0, (int) $options['time_limit_ms'] )
: max( 0, (int) ( $options['time_limit'] ?? 0 ) ) * 1000;
$stop_before_ms = isset( $options['stop_before_timeout_ms'] )
? max( 0, (int) $options['stop_before_timeout_ms'] )
: max( 0, (int) ( $options['stop_before_timeout'] ?? 0 ) ) * 1000;
$allow_group_fallback = ! isset( $options['allow_group_fallback'] ) || (bool) $options['allow_group_fallback'];
$execution_context = (string) ( $options['execution_context'] ?? 'agents-api scoped drain' );
$terminal_callback = is_callable( $options['terminal_status_callback'] ?? null ) ? $options['terminal_status_callback'] : null;
// SAFETY GATE 1: Action Scheduler's in-process runner must be present. When it
// is not, this is a clean no-op — the caller's run drains through whatever
// external pump the runtime does provide (or stays suspended until its budget),
// exactly as before this class existed. Never fabricate progress.
if ( ! self::is_available() ) {
return $this->empty_stats( 'as_unavailable', $hooks, $group );
}
// SAFETY GATE 2: refuse to run from a branch-worker / re-entrant context. This
// is the self-deadlock guard (see the class docblock): draining the scope from
// inside a claimed action of that same scope would hold a claim while pumping
// the queue its siblings need claims from. `doing_action()` on any scoped hook
// means this very request is a claimed scoped action; the static flag catches a
// nested drain within one request. Either way, refuse rather than deadlock.
$refusal = $this->guard_context( $hooks );
if ( '' !== $refusal ) {
return $this->empty_stats( $refusal, $hooks, $group );
}
self::$draining = true;
self::ensure_memory_limit();
$started_at = microtime( true );
$before_counts = $this->status_counts( $hooks, $group );
$batches = 0;
$processed = 0;
$warnings = 0;
$stop_reason = 'empty';
$terminal = '';
$diagnostic = null;
try {
while ( $this->due_pending_count( $hooks, $group ) > 0 ) {
// Stop the instant the run is done — do not waste a batch after
// completion. The callback returns a non-empty terminal state (e.g.
// 'succeeded' / 'failed') when the run is no longer suspended.
if ( null !== $terminal_callback ) {
$terminal = $this->scalar_string( $terminal_callback() );
if ( '' !== $terminal ) {
$stop_reason = 'terminal_status';
break;
}
}
if ( $this->memory_soft_limit_reached() ) {
$stop_reason = 'memory_limit';
break;
}
$elapsed_ms = (int) round( ( microtime( true ) - $started_at ) * 1000 );
if ( $time_limit_ms > 0 && $elapsed_ms >= $time_limit_ms ) {
$stop_reason = 'time_limit';
break;
}
if ( $time_limit_ms > 0 && ( $time_limit_ms - $elapsed_ms ) <= $stop_before_ms ) {
$stop_reason = 'timeout_margin';
break;
}
$current_batch = $batch_size;
if ( $limit > 0 ) {
$current_batch = min( $batch_size, $limit - $processed );
if ( $current_batch <= 0 ) {
$stop_reason = 'limit';
break;
}
}
$deadline_at = 0.0;
if ( $time_limit_ms > 0 ) {
$deadline_at = $started_at + max( 0, $time_limit_ms - $stop_before_ms ) / 1000;
}
$due_before = $this->due_pending_count( $hooks, $group );
$batch = $this->run_batch( $current_batch, $hooks, $group, $deadline_at, $execution_context, $allow_group_fallback );
++$batches;
$processed += (int) $batch['processed'];
$warnings += (int) $batch['warnings'];
if ( isset( $batch['diagnostic'] ) ) {
$diagnostic = $batch['diagnostic'];
}
if ( '' !== (string) $batch['stop_reason'] ) {
$stop_reason = (string) $batch['stop_reason'];
break;
}
// No observable progress AND the due set did not shrink → the scope is
// stuck (a claim we cannot make, a perpetually-rescheduled action). Stop
// rather than spin the loop against it.
if ( 0 === (int) $batch['processed'] && $this->due_pending_count( $hooks, $group ) >= $due_before ) {
$stop_reason = 'no_progress';
break;
}
}
// A run that reached terminal exactly as the scope emptied still reports the
// terminal stop reason (more informative than a bare 'empty').
if ( null !== $terminal_callback && '' === $terminal ) {
$terminal = $this->scalar_string( $terminal_callback() );
if ( '' !== $terminal && 'empty' === $stop_reason ) {
$stop_reason = 'terminal_status';
}
}
} finally {
self::$draining = false;
}
return $this->build_stats( $before_counts, $this->status_counts( $hooks, $group ), $hooks, $group, $batches, $processed, $warnings, $stop_reason, $terminal, $diagnostic );
}
/**
* Drain this executor's branch/resume scope until a suspended workflow run leaves
* the suspended state, or the supplied drain budget is exhausted.
*
* This is the generic foreground-await contract layered over {@see drain()}: a
* caller that already has the run id and recorder can pump its OWN suspended
* Action Scheduler fan-out without knowing the branch/resume hook names or
* reimplementing the terminal-status callback. Product code still owns WHEN it is
* safe to call this (foreground/status/CLI context, never from a branch worker).
*
* @since 0.5.2
*
* @param string $run_id Suspended workflow run id.
* @param WP_Agent_Workflow_Run_Recorder $recorder Recorder that can reload the run.
* @param array<string,mixed> $options Drain options forwarded to {@see drain()}.
* @return array{result:?WP_Agent_Workflow_Run_Result,stats:array<string,int|string|bool|array{code:string,message:string,actionable:bool}>}
*/
public function drain_suspended_run( string $run_id, WP_Agent_Workflow_Run_Recorder $recorder, array $options = array() ): array {
$terminal_status_callback = static function () use ( $recorder, $run_id ): string {
$current = $recorder->find( $run_id );
if ( null === $current ) {
return 'gone';
}
return $current->is_suspended() ? '' : $current->get_status();
};
$drain_options = array(
'terminal_status_callback' => $terminal_status_callback,
);
if ( isset( $options['hooks'] ) && is_array( $options['hooks'] ) ) {
$drain_options['hooks'] = array_values(
array_filter(
$options['hooks'],
static function ( $hook ): bool {
return is_string( $hook );
}
)
);
}
if ( isset( $options['group'] ) && is_scalar( $options['group'] ) && '' !== (string) $options['group'] ) {
$drain_options['group'] = (string) $options['group'];
}
foreach ( array( 'batch_size', 'limit', 'time_limit_ms', 'time_limit', 'stop_before_timeout_ms', 'stop_before_timeout' ) as $int_key ) {
if ( isset( $options[ $int_key ] ) && is_numeric( $options[ $int_key ] ) ) {
$drain_options[ $int_key ] = (int) $options[ $int_key ];
}
}
if ( isset( $options['allow_group_fallback'] ) ) {
$drain_options['allow_group_fallback'] = (bool) $options['allow_group_fallback'];
}
if ( isset( $options['execution_context'] ) && is_scalar( $options['execution_context'] ) ) {
$drain_options['execution_context'] = (string) $options['execution_context'];
}
$stats = $this->drain( $drain_options );
return array(
'result' => $recorder->find( $run_id ),
'stats' => $stats,
);
}
/**
* Claim and run one batch of due scoped actions in-process — pure Action
* Scheduler mechanics. Resets stale timeouts first, stakes a claim over the
* scoped hooks + group, then runs each claimed action through AS's own runner.
*
* @since 0.5.2
*
* @param int $batch_size Maximum actions to run this batch.
* @param array<int,string> $hooks Hook scope (null-equivalent handled by caller).
* @param string $group AS group.
* @param float $deadline_at Unix timestamp (with microseconds); 0 = no deadline.
* @param string $execution_context AS execution context label.
* @param bool $allow_group_fallback Whether HybridStore may retry without the group.
* @return array{processed:int,warnings:int,stop_reason:string,diagnostic?:array{code:string,message:string,actionable:bool}} Batch result.
*/
private function run_batch( int $batch_size, array $hooks, string $group, float $deadline_at, string $execution_context, bool $allow_group_fallback ): array {
/** @var \ActionScheduler_Store $store */
$store = \ActionScheduler_Store::instance();
$runner = \ActionScheduler::runner();
$processed = 0;
$warnings = 0;
$stop_reason = '';
$claim = null;
try {
$this->reset_stale_timeouts( $store );
// stake_claim( $max, $before_date, $hooks, $group ). Claiming over the
// scoped hooks + group means we only ever run THIS caller's scope — never
// an unrelated AS workload sharing the queue. HybridStore can transiently
// throw "group does not exist" while actions for that group are readable.
// Generic callers may retry hook-scoped; run awaiters disable that fallback
// because hook-only claims would cross run boundaries.
$claim = $this->stake_claim( $store, $batch_size, $hooks, $group, $allow_group_fallback );
} catch ( \Throwable $throwable ) {
return array(
'processed' => 0,
'warnings' => 1,
'stop_reason' => 'warning',
'diagnostic' => array(
'code' => 'scoped_drain_claim_failed',
'message' => self::safe_diagnostic_message( $throwable ),
'actionable' => true,
),
);
}
try {
foreach ( $claim->get_actions() as $action_id ) {
if ( $deadline_at > 0 && microtime( true ) >= $deadline_at ) {
$stop_reason = 'timeout_margin';
break;
}
if ( $this->memory_soft_limit_reached() ) {
$stop_reason = 'memory_limit';
break;
}
$action_id = (int) $action_id;
if ( $action_id <= 0 ) {
continue;
}
try {
$runner->process_action( $action_id, $execution_context );
++$processed;
} catch ( \Throwable $throwable ) {
unset( $throwable );
++$warnings;
} finally {
$this->flush_runtime_cache();
}
if ( $processed >= $batch_size ) {
break;
}
}
} finally {
$store->release_claim( $claim );
}
return array(
'processed' => $processed,
'warnings' => $warnings,
'stop_reason' => $stop_reason,
);
}
/**
* Stake a scoped claim, falling back to hook-only scope when HybridStore cannot
* resolve an otherwise-readable group.
*
* @since 0.5.2
*
* @param \ActionScheduler_Store $store Action Scheduler store.
* @param int $batch_size Maximum actions to claim.
* @param array<int,string> $hooks Hook scope.
* @param string $group Group scope.
* @param bool $allow_group_fallback Whether to retry hook-only.
* @return \ActionScheduler_ActionClaim Action Scheduler claim.
*/
private function stake_claim( \ActionScheduler_Store $store, int $batch_size, array $hooks, string $group, bool $allow_group_fallback ): \ActionScheduler_ActionClaim {
try {
return $store->stake_claim( $batch_size, null, $hooks, $group );
} catch ( \InvalidArgumentException $error ) {
if ( ! $allow_group_fallback || '' === $group || false === stripos( $error->getMessage(), 'group' ) ) {
throw $error;
}
return $store->stake_claim( $batch_size, null, $hooks, '' );
}
}
/**
* Refuse the drain when it is invoked from a claimed-action or re-entrant
* context. Returns a non-empty refusal reason string when the drain must NOT
* run, or '' when the foreground/orchestrator context is safe.
*
* @since 0.5.2
*
* @param array<int,string> $hooks Scoped hooks.
* @return string '' when safe, else the refusal stop_reason.
*/
private function guard_context( array $hooks ): string {
if ( self::$draining ) {
return 'refused_reentrant';
}
// If WordPress is currently firing one of the scoped hooks, THIS request is a
// claimed branch/resume action of the very scope we would drain — the
// self-deadlock case. Refuse. `doing_action()` is present in any WP context
// and reads the live action stack, so this is the exact, generic signal.
if ( function_exists( 'doing_action' ) ) {
foreach ( $hooks as $hook ) {
if ( doing_action( $hook ) ) {
return 'refused_in_claimed_action';
}
}
}
return '';
}
/**
* Reset stale AS claims/running actions before staking a direct drain claim, so a
* previously-abandoned claim does not hide a due action from this drain.
*
* @since 0.5.2
*
* @param object $store Action Scheduler store.
*/
private function reset_stale_timeouts( object $store ): void {
if ( ! class_exists( '\ActionScheduler_QueueCleaner' ) || ! ( $store instanceof \ActionScheduler_Store ) ) {
return;
}
$time_limit_raw = apply_filters( 'action_scheduler_queue_runner_time_limit', 30 );
$time_limit = is_numeric( $time_limit_raw ) ? (int) $time_limit_raw : 30;
$timeout = max( 1, $time_limit ) * 10;
$cleaner = new \ActionScheduler_QueueCleaner( $store );
$cleaner->reset_timeouts( $timeout );
$cleaner->mark_failures( $timeout );
}
/**
* Count due pending actions in scope (scheduled_date <= now).
*
* @since 0.5.2
*
* @param array<int,string> $hooks Scoped hooks.
* @param string $group AS group.
*/
private function due_pending_count( array $hooks, string $group ): int {
return $this->count_actions( $hooks, $group, true );
}
/**
* Count pending actions in scope, optionally only those that are due now.
*
* Sums per-hook because `as_get_scheduled_actions` filters by a single hook; the
* scope is a small hook list, so this is a handful of bounded indexed queries.
*
* @since 0.5.2
*
* @param array<int,string> $hooks Scoped hooks.
* @param string $group AS group.
* @param bool $due_only Whether to count only due actions.
*/
private function count_actions( array $hooks, string $group, bool $due_only ): int {
if ( ! function_exists( 'as_get_scheduled_actions' ) || ! class_exists( '\ActionScheduler_Store' ) ) {
return 0;
}
$total = 0;
foreach ( $hooks as $hook ) {
$query = array(
'hook' => $hook,
'group' => $group,
'status' => \ActionScheduler_Store::STATUS_PENDING,
'per_page' => 1000,
);
if ( $due_only ) {
// Only actions whose scheduled time has arrived are claimable now.
$query['date'] = as_get_datetime_object();
$query['date_compare'] = '<=';
}
$ids = as_get_scheduled_actions( $query, 'ids' );
$total += is_array( $ids ) ? count( $ids ) : 0;
}
return $total;
}
/**
* Get action counts grouped by hook and status for the scope — used to build the
* before/after processed deltas the stats report.
*
* @since 0.5.2
*
* @param array<int,string> $hooks Scoped hooks.
* @param string $group AS group.
* @return array<string,array<string,int>> Counts by hook and status.
*/
private function status_counts( array $hooks, string $group ): array {
$counts = array();
if ( ! function_exists( 'as_get_scheduled_actions' ) || ! class_exists( '\ActionScheduler_Store' ) ) {
return $counts;
}
$statuses = array(
'pending' => \ActionScheduler_Store::STATUS_PENDING,
'complete' => \ActionScheduler_Store::STATUS_COMPLETE,
'failed' => \ActionScheduler_Store::STATUS_FAILED,
);
foreach ( $hooks as $hook ) {
$counts[ $hook ] = array(
'pending' => 0,
'complete' => 0,
'failed' => 0,
);
foreach ( $statuses as $key => $as_status ) {
$ids = as_get_scheduled_actions(
array(
'hook' => $hook,
'group' => $group,
'status' => $as_status,
'per_page' => 1000,
),
'ids'
);
$counts[ $hook ][ $key ] = is_array( $ids ) ? count( $ids ) : 0;
}
}
return $counts;
}
/**
* Build operator-facing drain stats from before/after status snapshots.
*
* @since 0.5.2
*
* @param array<string,array<string,int>> $before Counts before.
* @param array<string,array<string,int>> $after Counts after.
* @param array<int,string> $hooks Scoped hooks.
* @param string $group AS group.
* @param int $batches Batches run.
* @param int $processed Actions processed.
* @param int $warnings Warnings.
* @param string $stop_reason Why the loop stopped.
* @param string $terminal Terminal state from the callback.
* @param array{code:string,message:string,actionable:bool}|null $diagnostic Safe claim diagnostic.
* @return array<string,int|string|bool|array{code:string,message:string,actionable:bool}> Stats.
*/
private function build_stats( array $before, array $after, array $hooks, string $group, int $batches, int $processed, int $warnings, string $stop_reason, string $terminal, ?array $diagnostic = null ): array {
$completions = 0;
$failures = 0;
foreach ( $hooks as $hook ) {
$completions += max( 0, ( $after[ $hook ]['complete'] ?? 0 ) - ( $before[ $hook ]['complete'] ?? 0 ) );
$failures += max( 0, ( $after[ $hook ]['failed'] ?? 0 ) - ( $before[ $hook ]['failed'] ?? 0 ) );
}
$stats = array(
'batches' => $batches,
'actions_processed' => $processed,
'completions' => $completions,
'failures' => $failures,
'remaining_pending' => $this->due_pending_count( $hooks, $group ),
'total_pending' => $this->count_actions( $hooks, $group, false ),
'warnings' => $warnings,
'stop_reason' => $stop_reason,
'terminal_state' => $terminal,
'hooks' => implode( ',', $hooks ),
'group' => $group,
'available' => true,
);
if ( null !== $diagnostic ) {
$stats['diagnostic'] = $diagnostic;
}
return $stats;
}
/**
* A stats shape for a drain that never ran a batch (unavailable / refused).
*
* @since 0.5.2
*
* @param string $stop_reason Why no batch ran.
* @param array<int,string> $hooks Scoped hooks.
* @param string $group AS group.
* @return array<string,int|string|bool|array{code:string,message:string,actionable:bool}> Stats.
*/
private function empty_stats( string $stop_reason, array $hooks, string $group ): array {
$available = self::is_available();
return array(
'batches' => 0,
'actions_processed' => 0,
'completions' => 0,
'failures' => 0,
'remaining_pending' => $available ? $this->due_pending_count( $hooks, $group ) : 0,
'total_pending' => $available ? $this->count_actions( $hooks, $group, false ) : 0,
'warnings' => 0,
'stop_reason' => $stop_reason,
'terminal_state' => '',
'hooks' => implode( ',', $hooks ),
'group' => $group,
'available' => $available,
);
}
/**
* Normalize the caller's hook scope, defaulting to the executor's branch + resume
* hooks. A caller may narrow or widen the scope, but never to an empty set.
*
* @since 0.5.2
*
* @param mixed $hooks Optional hook list.
* @return array<int,string> Non-empty hook scope.
*/
private function normalize_hooks( $hooks ): array {
if ( ! is_array( $hooks ) ) {
return self::default_hooks();
}
$normalized = array();
foreach ( $hooks as $hook ) {
$hook = is_string( $hook ) ? trim( $hook ) : '';
if ( '' !== $hook ) {
$normalized[] = $hook;
}
}
$normalized = array_values( array_unique( $normalized ) );
return empty( $normalized ) ? self::default_hooks() : $normalized;
}
/**
* Coerce an opaque callback return (the terminal-status callback returns mixed)
* to a string terminal state, treating non-scalars as "no terminal state yet".
*
* @since 0.5.2
*
* @param mixed $value Callback return.
*/
private function scalar_string( $value ): string {
if ( is_scalar( $value ) || $value instanceof \Stringable ) {
return (string) $value;
}
return '';
}
/** Return a stable summary without exposing arbitrary exception text to callers. */
private static function safe_diagnostic_message( \Throwable $error ): string {
return $error instanceof \InvalidArgumentException
? 'Action Scheduler could not claim the workflow action group.'
: 'Action Scheduler could not claim workflow actions.';
}
/**
* Flush in-request object cache state after each drained action so a long drain
* does not accumulate per-action cache growth.
*
* @since 0.5.2
*/
private function flush_runtime_cache(): void {
if ( function_exists( 'wp_cache_flush_runtime' )
&& ( ! function_exists( 'wp_cache_supports' ) || wp_cache_supports( 'flush_runtime' ) ) ) {
wp_cache_flush_runtime();
}
}
/**
* Whether PHP memory usage is near the hard limit. Filterable ratio via a
* generic `agents_`-prefixed filter so no product-specific filter name leaks
* into the substrate.
*
* @since 0.5.2
*/
private function memory_soft_limit_reached(): bool {
$limit = self::memory_limit_bytes();
if ( $limit <= 0 ) {
return false;
}
/**
* The fraction of the PHP memory limit at which the scoped drain stops before
* running the next action, so a long drain does not OOM.
*
* @since 0.5.2
*
* @param float $ratio Soft-limit ratio (clamped to 0.50–0.95). Default 0.80.
*/
$ratio = (float) apply_filters( 'agents_workflow_scoped_drain_memory_soft_limit_ratio', 0.80 );
$ratio = max( 0.50, min( 0.95, $ratio ) );
return memory_get_usage( true ) >= (int) floor( $limit * $ratio );
}
/**
* Raise the runtime memory floor for a large drain, mirroring Action Scheduler's
* own runner which raises to the admin memory limit before a batch.
*
* @since 0.5.2
*/
private static function ensure_memory_limit(): void {
if ( function_exists( 'wp_raise_memory_limit' ) ) {
wp_raise_memory_limit( 'admin' );
}
}
/**
* Return PHP memory_limit in bytes, or 0 for unlimited/unknown.
*
* @since 0.5.2
*/
private static function memory_limit_bytes(): int {
$raw = trim( (string) ini_get( 'memory_limit' ) );
if ( '' === $raw || '-1' === $raw ) {
return 0;
}
$unit = strtolower( substr( $raw, -1 ) );
$value = (float) $raw;
switch ( $unit ) {
case 'g':
$value *= 1024;
// Fall through.
case 'm':
$value *= 1024;
// Fall through.
case 'k':
$value *= 1024;
}
return max( 0, (int) $value );
}
}