Skip to content

Expose pipeline outcome classes #2173

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 2 commits into from
May 22, 2026
Merged

Expose pipeline outcome classes #2173

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fcfd0c3
fix: expose pipeline outcome classes
chubes4 May 22, 2026
099b66b
fix: align outcome class assignments
chubes4 May 22, 2026
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
7 changes: 6 additions & 1 deletion inc/Abilities/Engine/ExecuteStepAbility.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,7 @@ public function execute( array $input ): array {
'step_success' => $step_success,
'packet_count' => count( $dataPackets ),
'status' => $recorded_status,
'reason' => $result['reason'] ?? ( $result['error'] ?? null ),
'error' => $result['error'] ?? null,
)
);
Expand Down Expand Up @@ -595,6 +596,7 @@ private function routeAfterExecution(
'success' => true,
'step_success' => false,
'outcome' => 'failed',
'reason' => $transition_failure_reason,
'error' => $transition_failure_reason,
);
}
Expand Down Expand Up @@ -742,21 +744,24 @@ private function routeAfterExecution(
'step_type' => $step_type,
)
);
$empty_packet_reason = $this->getFailureReasonFromPackets( $dataPackets, 'empty_data_packet_returned' );

do_action(
'datamachine_fail_job',
$job_id,
'step_execution_failure',
array(
'flow_step_id' => $flow_step_id,
'class' => $step_class,
'reason' => $this->getFailureReasonFromPackets( $dataPackets, 'empty_data_packet_returned' ),
'reason' => $empty_packet_reason,
)
);

return array(
'success' => true,
'step_success' => false,
'outcome' => 'failed',
'reason' => $empty_packet_reason,
);
}

Expand Down
5 changes: 5 additions & 0 deletions inc/Cli/Commands/JobsCommand.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1777,6 +1777,7 @@ public function metrics( array $args, array $assoc_args ): void {
$counts = $metrics['counts'] ?? array();
$children = $metrics['child_jobs'] ?? array();
$timestamps = $metrics['timestamps'] ?? array();
$classes = is_array( $metrics['outcome_classes'] ?? null ) ? $metrics['outcome_classes'] : array();

WP_CLI::log( sprintf( 'Job ID: %d', $metrics['job_id'] ?? 0 ) );
WP_CLI::log( sprintf( 'Status: %s', $metrics['status'] ?? '' ) );
Expand All @@ -1796,6 +1797,10 @@ public function metrics( array $args, array $assoc_args ): void {
}
WP_CLI::log( '' );

WP_CLI::log( 'Outcome Classes:' );
WP_CLI::log( empty( $classes ) ? ' -' : ' ' . implode( ', ', array_map( 'strval', $classes ) ) );
WP_CLI::log( '' );

WP_CLI::log( 'Child Jobs:' );
foreach ( $children as $key => $value ) {
WP_CLI::log( sprintf( ' %s: %d', $key, (int) $value ) );
Expand Down
126 changes: 124 additions & 2 deletions inc/Core/RunMetrics.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,14 @@ class RunMetrics {
'failed',
'fetch_packets',
'no_content',
'true_empty_query',
'provider_error',
'hydration_failed',
'hydration_partial',
'ai_empty_packet',
'missing_handler_packet',
'source_rejected',
'item_deferred',
'retried',
'scheduled',
'staged_actions',
Expand Down Expand Up @@ -66,6 +73,12 @@ public static function recordStepResult( int $job_id, string $flow_step_id, arra
if ( 'source_rejected' === ( $clean_result['result'] ?? '' ) || ! empty( $clean_result['source_rejection_reason'] ) ) {
$metrics['counts']['source_rejected'] = max( 1, (int) $metrics['counts']['source_rejected'] );
}
foreach ( self::classesFromStepResult( $clean_result, (string) ( $clean_result['status'] ?? '' ) ) as $class ) {
if ( ! isset( $metrics['counts'][ $class ] ) ) {
$metrics['counts'][ $class ] = 0;
}
$metrics['counts'][ $class ] = max( 1, (int) $metrics['counts'][ $class ] );
}
$metrics['last_activity_at'] = self::now();
$engine[ self::KEY ] = $metrics;

Expand Down Expand Up @@ -161,6 +174,8 @@ public static function fromJob( array $job ): array {
$counts[ $key ] = max( (int) ( $counts[ $key ] ?? 0 ), (int) $value );
}

$outcome_classes = self::outcomeClasses( $job, $engine, $counts );

return array(
'job_id' => (int) ( $job['job_id'] ?? 0 ),
'source' => $job['source'] ?? null,
Expand All @@ -170,6 +185,7 @@ public static function fromJob( array $job ): array {
'parent_job_id' => isset( $job['parent_job_id'] ) ? (int) $job['parent_job_id'] : 0,
'status' => $status,
'counts' => $counts,
'outcome_classes' => $outcome_classes,
'child_jobs' => self::childTotals( (int) ( $job['job_id'] ?? 0 ) ),
'timestamps' => array(
'created_at' => $job['created_at'] ?? null,
Expand All @@ -178,7 +194,7 @@ public static function fromJob( array $job ): array {
'completed_at' => $ended_at,
),
'duration_seconds' => self::durationSeconds( $started_at, $duration_end ),
'outcome' => self::outcomeDetails( $job, $engine, $counts ),
'outcome' => self::outcomeDetails( $job, $engine, $counts, $outcome_classes ),
'step_results' => self::stepResults( $engine ),
'context' => $metrics['context'],
'token_usage' => self::tokenUsage( $engine ),
Expand Down Expand Up @@ -270,22 +286,36 @@ private static function inferCountsFromEngine( array $engine, string $status ):
if ( 'source_rejected' === ( $step_result['result'] ?? '' ) || ! empty( $step_result['source_rejection_reason'] ) ) {
$counts['source_rejected'] = max( 1, (int) ( $counts['source_rejected'] ?? 0 ) );
}
foreach ( self::classesFromStepResult( $step_result, $status ) as $class ) {
$counts[ $class ] = max( 1, (int) ( $counts[ $class ] ?? 0 ) );
}
}

foreach ( self::classesFromStatus( $status ) as $class ) {
$counts[ $class ] = max( 1, (int) ( $counts[ $class ] ?? 0 ) );
}

return $counts;
}

private static function outcomeDetails( array $job, array $engine, array $counts ): array {
private static function outcomeDetails( array $job, array $engine, array $counts, array $outcome_classes ): array {
$status = (string) ( $job['status'] ?? '' );
$job_status = JobStatus::fromString( $status );
$source_rejection = is_array( $engine['source_rejection'] ?? null ) ? $engine['source_rejection'] : array();
$is_source_rejected = ! empty( $counts['source_rejected'] ) || ! empty( $source_rejection ) || 'source-rejected' === $job_status->getReason();
$class_counts = array();
foreach ( $outcome_classes as $class ) {
$class_counts[ $class ] = (int) ( $counts[ $class ] ?? 0 );
}

return array_filter(
array(
'status' => $status,
'base_status' => $job_status->getBaseStatus(),
'status_reason' => $job_status->getReason(),
'primary_class' => $outcome_classes[0] ?? null,
'classes' => $outcome_classes,
'class_counts' => $class_counts,
'fetch_packet_count' => (int) ( $counts['fetch_packets'] ?? 0 ),
'no_content' => ! empty( $counts['no_content'] ) || JobStatus::COMPLETED_NO_ITEMS === $job_status->getBaseStatus(),
'source_rejected' => $is_source_rejected,
Expand All @@ -298,6 +328,98 @@ private static function outcomeDetails( array $job, array $engine, array $counts
);
}

private static function outcomeClasses( array $job, array $engine, array $counts ): array {
$status = (string) ( $job['status'] ?? '' );
$classes = self::classesFromStatus( $status );

foreach ( self::stepResults( $engine ) as $step_result ) {
$classes = array_merge( $classes, self::classesFromStepResult( $step_result, $status ) );
}

foreach ( array( 'true_empty_query', 'provider_error', 'hydration_failed', 'hydration_partial', 'ai_empty_packet', 'missing_handler_packet', 'source_rejected', 'item_deferred' ) as $class ) {
if ( ! empty( $counts[ $class ] ) ) {
$classes[] = $class;
}
}

return array_values( array_unique( array_filter( $classes ) ) );
}

private static function classesFromStepResult( array $step_result, string $status = '' ): array {
$result = (string) ( $step_result['result'] ?? '' );
$reason = self::outcomeReasonFrom( $step_result, $status );
$classes = self::classesFromReason( $reason );

if ( in_array( $result, array( 'no_content', 'completed_no_items' ), true ) ) {
$classes[] = 'true_empty_query';
}
if ( 'source_rejected' === $result || ! empty( $step_result['source_rejection_reason'] ) ) {
$classes[] = 'source_rejected';
}
if ( 'fetch' === ( $step_result['step_type'] ?? '' ) && 'failed' === $result ) {
$classes[] = 'provider_error';
}

return array_values( array_unique( array_filter( $classes ) ) );
}

private static function classesFromStatus( string $status ): array {
$job_status = JobStatus::fromString( $status );
$classes = self::classesFromReason( (string) $job_status->getReason() );

if ( JobStatus::COMPLETED_NO_ITEMS === $job_status->getBaseStatus() ) {
$classes[] = 'true_empty_query';
}

return array_values( array_unique( array_filter( $classes ) ) );
}

private static function classesFromReason( string $reason ): array {
$reason = strtolower( str_replace( '-', '_', trim( $reason ) ) );
if ( '' === $reason ) {
return array();
}

$classes = array();
if ( in_array( $reason, array( 'mcp_fetch_failed', 'auth_ref_resolution_failed', 'ai_provider_missing' ), true ) || str_contains( $reason, 'provider' ) ) {
$classes[] = 'provider_error';
}
if ( 'missing_source_content' === $reason || str_contains( $reason, 'hydration_failed' ) ) {
$classes[] = 'hydration_failed';
}
if ( str_contains( $reason, 'hydration_partial' ) ) {
$classes[] = 'hydration_partial';
}
if ( 'empty_data_packet_returned' === $reason ) {
$classes[] = 'ai_empty_packet';
}
if ( 'handler_requiring_step_missing_handler_packets' === $reason ) {
$classes[] = 'missing_handler_packet';
}
if ( 'source_rejected' === $reason ) {
$classes[] = 'source_rejected';
}
if ( 'item_deferred' === $reason ) {
$classes[] = 'item_deferred';
}

return $classes;
}

private static function outcomeReasonFrom( array $step_result, string $status ): string {
foreach ( array( 'reason', 'error', 'status' ) as $key ) {
$value = $step_result[ $key ] ?? null;
if ( is_scalar( $value ) && '' !== (string) $value ) {
if ( 'status' === $key ) {
return (string) JobStatus::fromString( (string) $value )->getReason();
}
return (string) $value;
}
}

return (string) JobStatus::fromString( $status )->getReason();
}

private static function stepResults( array $engine ): array {
$step_results = is_array( $engine[ self::STEP_RESULTS_KEY ] ?? null ) ? $engine[ self::STEP_RESULTS_KEY ] : array();
return array_values( array_filter( $step_results, 'is_array' ) );
Expand Down
15 changes: 15 additions & 0 deletions inc/Core/Steps/Fetch/Tools/FetchItemDispositionTool.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,21 @@ private function deferItem( array $parameters, array $tool_def ): array {
)
);

if ( $flow_step_id && class_exists( RunMetrics::class ) ) {
RunMetrics::recordStepResult(
$job_id,
(string) $flow_step_id,
array(
'step_type' => 'fetch',
'result' => 'item_deferred',
'packet_count' => 0,
'reason' => 'item-deferred',
'source_type' => $source_type,
'item_identifier' => $item_identifier,
)
);
}

do_action(
'datamachine_log',
'info',
Expand Down
53 changes: 52 additions & 1 deletion tests/job-outcome-metrics-smoke.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,8 @@
);
$assert( 'base status is completed_no_items', 'completed_no_items' === $metrics['outcome']['base_status'] );
$assert( 'no_content boolean is true', true === $metrics['outcome']['no_content'] );
$assert( 'true empty query class is exposed', in_array( 'true_empty_query', $metrics['outcome_classes'], true ) );
$assert( 'true empty query count is exposed', 1 === $metrics['counts']['true_empty_query'] );

echo "\n[3] failed status exposes failure base status\n";
$metrics = RunMetrics::fromJob( $job( 'failed - api-timeout', array() ) );
Expand All @@ -113,14 +115,63 @@
);
$assert( 'source_rejected boolean is true', true === $metrics['outcome']['source_rejected'] );
$assert( 'source rejection reason is exposed', 'not relevant' === $metrics['outcome']['source_rejection_reason'] );
$assert( 'source rejected outcome class is exposed', array( 'source_rejected' ) === $metrics['outcome_classes'] );

echo "\n[5] CLI/source integration markers exist\n";
echo "\n[5] failure reasons expose distinct generic outcome classes\n";
$metrics = RunMetrics::fromJob( $job( 'failed - mcp_fetch_failed', array() ) );
$assert( 'provider failure class is exposed from status reason', array( 'provider_error' ) === $metrics['outcome_classes'] );
$assert( 'provider failure count is exposed', 1 === $metrics['counts']['provider_error'] );

$metrics = RunMetrics::fromJob( $job( 'failed - missing_source_content', array() ) );
$assert( 'hydration failure class is exposed from status reason', array( 'hydration_failed' ) === $metrics['outcome_classes'] );

$metrics = RunMetrics::fromJob(
$job(
'failed - step_execution_failure',
array(
'step_results' => array(
'ai_1' => array(
'flow_step_id' => 'ai_1',
'step_type' => 'ai',
'result' => 'failed',
'reason' => 'empty_data_packet_returned',
'packet_count' => 0,
),
),
)
)
);
$assert( 'AI empty packet class is exposed from step result', array( 'ai_empty_packet' ) === $metrics['outcome_classes'] );

$metrics = RunMetrics::fromJob(
$job(
'failed - step_execution_failure',
array(
'step_results' => array(
'ai_1' => array(
'flow_step_id' => 'ai_1',
'step_type' => 'ai',
'result' => 'failed',
'reason' => 'handler_requiring_step_missing_handler_packets',
'packet_count' => 2,
),
),
)
)
);
$assert( 'missing handler packet class is exposed from step result', array( 'missing_handler_packet' ) === $metrics['outcome_classes'] );

$metrics = RunMetrics::fromJob( $job( 'failed - item-deferred', array() ) );
$assert( 'item deferred class is exposed from status reason', array( 'item_deferred' ) === $metrics['outcome_classes'] );

echo "\n[6] CLI/source integration markers exist\n";
$jobs_command = file_get_contents( __DIR__ . '/../inc/Cli/Commands/JobsCommand.php' ) ?: '';
$fetch_step = file_get_contents( __DIR__ . '/../inc/Core/Steps/Fetch/FetchStep.php' ) ?: '';
$disposition = file_get_contents( __DIR__ . '/../inc/Core/Steps/Fetch/Tools/FetchItemDispositionTool.php' ) ?: '';
$assert( 'jobs list supports pipeline filter', str_contains( $jobs_command, "assoc_args['pipeline']" ) );
$assert( 'jobs list supports handler filter', str_contains( $jobs_command, "assoc_args['handler']" ) );
$assert( 'jobs list JSON includes outcome', str_contains( $jobs_command, "item['outcome']" ) );
$assert( 'jobs metrics table prints outcome classes', str_contains( $jobs_command, 'Outcome Classes:' ) );
$assert( 'fetch step records packet count', str_contains( $fetch_step, "'packet_count' => count( \$packets )" ) );
$assert( 'source rejection persists structured reason', str_contains( $disposition, "'source_rejection'" ) );

Expand Down
Loading