Skip to content

Fix bounded cleanup run apply batches #551

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chubes4 merged 1 commit into from
Jun 4, 2026
Merged

Fix bounded cleanup run apply batches #551

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
20 changes: 18 additions & 2 deletions inc/Abilities/WorkspaceAbilities.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2228,6 +2228,7 @@ private function registerAbilities(): void {
'properties' => array(
'run_id' => array( 'type' => 'string' ),
'force' => array( 'type' => 'boolean' ),
'limit' => array( 'type' => 'integer' ),
),
),
'output_schema' => array( 'type' => 'object' ),
Expand All @@ -2250,6 +2251,7 @@ private function registerAbilities(): void {
'properties' => array(
'run_id' => array( 'type' => 'string' ),
'force' => array( 'type' => 'boolean' ),
'limit' => array( 'type' => 'integer' ),
),
),
'output_schema' => array( 'type' => 'object' ),
Expand Down Expand Up @@ -3489,7 +3491,7 @@ public static function workspaceCleanupPlan( array $input ): array|\WP_Error {
* @return array<string,mixed>|\WP_Error
*/
public static function workspaceCleanupApply( array $input ): array|\WP_Error {
return ( new CleanupRunService() )->apply( (string) ( $input['run_id'] ?? '' ), array( 'force' => ! empty($input['force']) ));
return ( new CleanupRunService() )->apply( (string) ( $input['run_id'] ?? '' ), self::cleanupRunApplyOptions($input));
}

/**
Expand Down Expand Up @@ -3519,7 +3521,21 @@ public static function workspaceCleanupEvidence( array $input ): array|\WP_Error
* @return array<string,mixed>|\WP_Error
*/
public static function workspaceCleanupResume( array $input ): array|\WP_Error {
return ( new CleanupRunService() )->resume( (string) ( $input['run_id'] ?? '' ), array( 'force' => ! empty($input['force']) ));
return ( new CleanupRunService() )->resume( (string) ( $input['run_id'] ?? '' ), self::cleanupRunApplyOptions($input));
}

/**
* Normalize bounded cleanup apply/resume options.
*
* @param array<string,mixed> $input Ability input.
* @return array<string,mixed>
*/
private static function cleanupRunApplyOptions( array $input ): array {
$options = array( 'force' => ! empty($input['force']) );
if ( isset($input['limit']) ) {
$options['limit'] = (int) $input['limit'];
}
return $options;
}

/**
Expand Down
12 changes: 7 additions & 5 deletions inc/Cli/Commands/WorkspaceCommand.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -327,10 +327,12 @@ public function adopt_repo( array $args, array $assoc_args ): void {
* : Pass an age gate such as 7d or 24h into cleanup task params.
*
* [--limit=<count>]
* : Maximum worktrees to scan in a `--mode=artifacts` page. Dry-run reviews
* scan this bounded page synchronously; apply runs freeze eligible candidates
* from the same bounded page and schedule only those candidates. Defaults to
* 100. Use 0 to disable the cap (combine with --exhaustive for a full audit).
* : For DB-backed `apply` / `resume`, maximum pending rows to process in this
* invocation (default 25, max 100). For `--mode=artifacts` pages, maximum
* worktrees to scan; dry-run reviews scan this bounded page synchronously,
* and apply runs freeze eligible candidates from the same bounded page.
* Artifact page scans default to 100. Use 0 to disable the artifact scan cap
* (combine with --exhaustive for a full audit).
*
* [--offset=<count>]
* : Pagination offset (0-indexed) for `--mode=artifacts` dry-run and apply
Expand Down Expand Up @@ -547,7 +549,7 @@ private function run_cleanup_control_ability( string $operation, string $run_id,
array(
'run_id' => $run_id,
'force' => ! empty($assoc_args['force']),
)
) + ( isset($assoc_args['limit']) ? array( 'limit' => (int) $assoc_args['limit'] ) : array() )
);
if ( is_wp_error($result) ) {
WP_CLI::error($result->get_error_message());
Expand Down
77 changes: 60 additions & 17 deletions inc/Workspace/CleanupRunService.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@

class CleanupRunService {

private const DEFAULT_APPLY_LIMIT = 25;
private const MAX_APPLY_LIMIT = 100;



public function __construct(
Expand Down Expand Up @@ -80,54 +83,89 @@ public function apply( string $run_id, array $opts = array() ): array|\WP_Error
return new \WP_Error('cleanup_run_not_found', sprintf('Cleanup run not found: %s', $run_id), array( 'status' => 404 ));
}

$limit = $this->apply_limit($opts);

$this->repository->update_run(
$run_id, array(
'status' => 'applying',
'started_at' => gmdate('Y-m-d H:i:s'),
)
);

$items = $this->repository->get_items($run_id);
$artifact_rows = $this->pending_rows_of_type($items, 'artifact_cleanup');
$worktree_rows = $this->pending_rows_of_type($items, 'worktree_removal');
$results = array();
$items = $this->repository->get_items($run_id);
$artifact_rows = $this->pending_rows_of_type($items, 'artifact_cleanup');
$worktree_rows = $this->pending_rows_of_type($items, 'worktree_removal');
$batch_type = '';
$processed_rows = 0;
$remaining_rows = max(0, count($artifact_rows) + count($worktree_rows));
$results = array();

if ( array() !== $artifact_rows ) {
$artifact_batch = array_slice($artifact_rows, 0, $limit);
$processed_rows += count($artifact_batch);
$batch_type = 'artifact_cleanup';
$results['artifact_cleanup'] = $this->workspace->worktree_cleanup_artifacts(
array(
'apply_plan' => array( 'candidates' => array_map(fn( $item ) => $item['evidence'], $artifact_rows) ),
'apply_plan' => array( 'candidates' => array_map(fn( $item ) => $item['evidence'], $artifact_batch) ),
'force' => ! empty($opts['force']),
'limit' => count($artifact_rows),
'limit' => count($artifact_batch),
)
);
$this->record_apply_result($artifact_rows, $results['artifact_cleanup'], 'removed');
$this->record_apply_result($artifact_batch, $results['artifact_cleanup'], 'removed');
}

if ( array() !== $worktree_rows ) {
$remaining_capacity = max(0, $limit - $processed_rows);
if ( $remaining_capacity > 0 && array() !== $worktree_rows ) {
$worktree_batch = array_slice($worktree_rows, 0, $remaining_capacity);
$processed_rows += count($worktree_batch);
$batch_type = '' === $batch_type ? 'worktree_removal' : 'mixed';
$results['worktree_removal'] = $this->workspace->worktree_cleanup_merged(
array(
'apply_plan' => array( 'candidates' => array_map(fn( $item ) => $item['evidence'], $worktree_rows) ),
'apply_plan' => array( 'candidates' => array_map(fn( $item ) => $item['evidence'], $worktree_batch) ),
'skip_github' => true,
)
);
$this->record_apply_result($worktree_rows, $results['worktree_removal'], 'removed');
$this->record_apply_result($worktree_batch, $results['worktree_removal'], 'removed');
}

$status = $this->status($run_id);
$summary = $status instanceof \WP_Error ? array() : ( $status['summary'] ?? array() );
$pending_or_fail = (int) ( $summary['pending_or_failed'] ?? 0 );
$next_status = $pending_or_fail > 0 ? 'needs_resume' : 'completed';
$completed_at = 'completed' === $next_status ? gmdate('Y-m-d H:i:s') : null;

$this->repository->update_run(
$run_id, array(
'status' => 'completed',
'completed_at' => gmdate('Y-m-d H:i:s'),
'summary' => $this->status($run_id)['summary'] ?? array(),
'status' => $next_status,
'completed_at' => $completed_at,
'summary' => $summary,
)
);

$status = $this->status($run_id);
if ( $status instanceof \WP_Error ) {
return $status;
}

return array(
'success' => true,
'state' => 'completed',
'state' => $next_status,
'run_id' => $run_id,
'status' => 'completed',
'results' => $results,
'summary' => $this->status($run_id)['summary'] ?? array(),
'status' => $next_status,
'batch' => array(
'type' => $batch_type,
'limit' => $limit,
'processed_rows' => $processed_rows,
'remaining_before' => $remaining_rows,
'remaining_after' => $pending_or_fail,
),
'results' => $results,
'summary' => $status['summary'] ?? array(),
'remaining_work_summary' => $status['remaining_work_summary'] ?? array(),
'next' => $pending_or_fail > 0 ? array(
'resume_command' => sprintf('studio wp datamachine-code workspace cleanup resume %s --limit=%d', $run_id, $limit),
'pending_rows' => $pending_or_fail,
) : null,
);
}

Expand Down Expand Up @@ -250,6 +288,11 @@ private function pending_rows_of_type( array $items, string $type ): array {
return array_values(array_filter($items, fn( $item ) => (string) ( $item['item_type'] ?? '' ) === $type && in_array( (string) ( $item['status'] ?? '' ), array( 'pending', 'failed' ), true)));
}

private function apply_limit( array $opts ): int {
$limit = isset($opts['limit']) ? (int) $opts['limit'] : self::DEFAULT_APPLY_LIMIT;
return max(1, min(self::MAX_APPLY_LIMIT, $limit));
}

private function record_apply_result( array $items, mixed $result, string $applied_key ): void {
if ( $result instanceof \WP_Error ) {
foreach ( $items as $item ) {
Expand Down
58 changes: 53 additions & 5 deletions tests/smoke-cleanup-run-storage.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,9 @@ function wp_json_encode( mixed $value, int $flags = 0 ): string|false
if (! function_exists('wp_generate_password') ) {
function wp_generate_password( int $length = 12 ): string
{
return substr(str_repeat('a', $length), 0, $length);
static $counter = 0;
++$counter;
return substr(str_repeat((string) $counter, $length), 0, $length);
}
}

Expand Down Expand Up @@ -125,6 +127,9 @@ public function get_results( string $query, string $output = ARRAY_A ): array

class DataMachineCodeCleanupRunFakeWorkspace extends \DataMachineCode\Workspace\Workspace
{
public array $artifact_calls = array();
public array $worktree_calls = array();

public function __construct()
{
}
Expand Down Expand Up @@ -156,17 +161,41 @@ public function workspace_cleanup_plan( array $opts = array() ): array|WP_Error
}
public function worktree_cleanup_artifacts( array $opts = array() ): array|WP_Error
{
$this->artifact_calls[] = $opts;
$candidates = (array) ( $opts['apply_plan']['candidates'] ?? array() );
$removed = array();
$skipped = array();
foreach ( $candidates as $candidate ) {
$handle = (string) ( is_array($candidate) ? ( $candidate['handle'] ?? '' ) : '' );
if ('demo@old' === $handle ) {
$removed[] = array( 'handle' => 'demo@old', 'artifact_size_bytes' => 15 );
} elseif ('demo@stale-artifact' === $handle ) {
$skipped[] = array( 'handle' => 'demo@stale-artifact', 'reason_code' => 'plan_mismatch', 'reason' => 'artifact plan no longer matches', 'artifact_size_bytes' => 9 );
}
}
return array(
'removed' => array( array( 'handle' => 'demo@old', 'artifact_size_bytes' => 15 ) ),
'skipped' => array( array( 'handle' => 'demo@stale-artifact', 'reason_code' => 'plan_mismatch', 'reason' => 'artifact plan no longer matches', 'artifact_size_bytes' => 9 ) ),
'removed' => $removed,
'skipped' => $skipped,
'summary' => array(),
);
}
public function worktree_cleanup_merged( array $opts = array() ): array|WP_Error
{
$this->worktree_calls[] = $opts;
$candidates = (array) ( $opts['apply_plan']['candidates'] ?? array() );
$removed = array();
$skipped = array();
foreach ( $candidates as $candidate ) {
$handle = (string) ( is_array($candidate) ? ( $candidate['handle'] ?? '' ) : '' );
if ('demo@merged' === $handle ) {
$removed[] = array( 'handle' => 'demo@merged', 'size_bytes' => 20 );
} elseif ('demo@dirty' === $handle ) {
$skipped[] = array( 'handle' => 'demo@dirty', 'reason_code' => 'dirty_worktree', 'reason' => 'working tree is dirty' );
}
}
return array(
'removed' => array( array( 'handle' => 'demo@merged', 'size_bytes' => 20 ) ),
'skipped' => array( array( 'handle' => 'demo@dirty', 'reason_code' => 'dirty_worktree', 'reason' => 'working tree is dirty' ) ),
'removed' => $removed,
'skipped' => $skipped,
'summary' => array(),
);
}
Expand Down Expand Up @@ -212,4 +241,23 @@ function datamachine_code_cleanup_run_assert( bool $condition, string $message )
datamachine_code_cleanup_run_assert(4 === (int) ( $evidence['remaining_work_summary']['blocked_resolvers_by_reason']['needs_metadata_reconcile']['count'] ?? 0 ), 'evidence keeps blocked resolver bucket');
datamachine_code_cleanup_run_assert(str_contains((string) wp_json_encode($evidence['remaining_work_summary']['recommended_commands']), 'workspace worktree reconcile-metadata'), 'summary recommends next DMC commands');

$bounded_workspace = new DataMachineCodeCleanupRunFakeWorkspace();
$bounded_service = new \DataMachineCode\Workspace\CleanupRunService($repo, $bounded_workspace);
$bounded_plan = $bounded_service->plan(array( 'mode' => 'retention' ));
datamachine_code_cleanup_run_assert(! is_wp_error($bounded_plan), 'bounded plan succeeds');

$bounded_apply = $bounded_service->apply($bounded_plan['run_id'], array( 'limit' => 1 ));
datamachine_code_cleanup_run_assert('needs_resume' === (string) ( $bounded_apply['status'] ?? '' ), 'bounded apply pauses when rows remain');
datamachine_code_cleanup_run_assert(1 === (int) ( $bounded_apply['batch']['processed_rows'] ?? 0 ), 'bounded apply processes one row');
datamachine_code_cleanup_run_assert(str_contains((string) ( $bounded_apply['next']['resume_command'] ?? '' ), 'workspace cleanup resume'), 'bounded apply returns resume command');
datamachine_code_cleanup_run_assert(1 === count($bounded_workspace->artifact_calls), 'bounded apply only runs artifact cleanup first');
datamachine_code_cleanup_run_assert(0 === count($bounded_workspace->worktree_calls), 'bounded apply defers worktree cleanup until artifacts drain');

$bounded_service->resume($bounded_plan['run_id'], array( 'limit' => 1 ));
$bounded_service->resume($bounded_plan['run_id'], array( 'limit' => 1 ));
$bounded_done = $bounded_service->resume($bounded_plan['run_id'], array( 'limit' => 1 ));
datamachine_code_cleanup_run_assert('completed' === (string) ( $bounded_done['status'] ?? '' ), 'bounded resume completes final batch');
datamachine_code_cleanup_run_assert(2 === count($bounded_workspace->artifact_calls), 'bounded resume drains artifact batches before worktrees');
datamachine_code_cleanup_run_assert(2 === count($bounded_workspace->worktree_calls), 'bounded resume drains worktree batches after artifacts');

echo "DB-backed cleanup run storage smoke passed.\n";
12 changes: 12 additions & 0 deletions tests/smoke-worktree-cleanup-cli.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -723,7 +723,9 @@ public function execute( array $input ): array
$fail_job_ability = new FakeFailJobAbility();
$GLOBALS['__abilities'] = array(
'datamachine-code/workspace-cleanup-run' => $cleanup_run_ability,
'datamachine-code/workspace-cleanup-apply' => $cleanup_status_ability,
'datamachine-code/workspace-cleanup-status' => $cleanup_status_ability,
'datamachine-code/workspace-cleanup-resume' => $cleanup_status_ability,
'datamachine-code/workspace-hygiene-report' => $hygiene_ability,
'datamachine-code/workspace-worktree-cleanup' => $ability,
'datamachine-code/workspace-worktree-cleanup-artifacts' => $artifact_ability,
Expand Down Expand Up @@ -802,6 +804,16 @@ public function execute( array $input ): array
datamachine_code_cleanup_assert('cleanup-run-20260504193024-abc123' === ( $cleanup_status_ability->last_input['run_id'] ?? '' ), 'DB cleanup run IDs are routed to cleanup status ability');
datamachine_code_cleanup_assert('planned' === ( $db_status_json['state'] ?? '' ), 'DB cleanup run status does not route to job-backed status parser');

WP_CLI::$logs = array();
WP_CLI::$successes = array();
$command->cleanup(array( 'apply', 'cleanup-run-20260504193024-abc123' ), array( 'limit' => 7, 'format' => 'json' ));
datamachine_code_cleanup_assert(7 === (int) ( $cleanup_status_ability->last_input['limit'] ?? 0 ), 'DB cleanup apply forwards bounded limit');

WP_CLI::$logs = array();
WP_CLI::$successes = array();
$command->cleanup(array( 'resume', 'cleanup-run-20260504193024-abc123' ), array( 'limit' => 9, 'format' => 'json' ));
datamachine_code_cleanup_assert(9 === (int) ( $cleanup_status_ability->last_input['limit'] ?? 0 ), 'DB cleanup resume forwards bounded limit');

WP_CLI::$logs = array();
WP_CLI::$successes = array();
$command->cleanup(array( 'status', 'cleanup-run-123' ), array( 'format' => 'json' ));
Expand Down
Loading