Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,10 @@ private function createChildJob(
// like CoreMemoryFilesDirective resolve the correct agent's
// MEMORY.md / SOUL.md instead of falling back to the user_id
// default-agent lookup.
$child_engine = $this->stripBatchRuntimeState( \DataMachine\Core\EngineData::stripFlowRuntimeQueuePayloads( $engine_snapshot ) );
// The parent was seeded with the first packet's engine data; withoutSeed()
// takes it back out so that packet's item fields cannot leak into children
// whose own packets leave those keys unset.
$child_engine = PacketEngineData::withoutSeed( $this->stripBatchRuntimeState( \DataMachine\Core\EngineData::stripFlowRuntimeQueuePayloads( $engine_snapshot ) ) );
unset( $child_engine[ ProcessedItems::CLAIM_METADATA_KEY ], $child_engine[ ProcessedItems::CLAIMS_METADATA_KEY ] );
$child_engine['job'] = array(
'job_id' => $child_job_id,
Expand Down
78 changes: 78 additions & 0 deletions plugins/data-machine/inc/Core/PacketEngineData.php
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,84 @@ public static function sanitize( array $engine_data, int $job_id ): array {
return $engine_data;
}

/**
* Engine key recording what a packet seed changed on a parent job.
*
* Shape: array{keys: string[], prior: array<string, mixed>}. `keys` are the
* top-level keys the seed set; `prior` holds the values those keys had
* before seeding (absent when the key did not exist).
*/
public const SEED_RESTORE_KEY = 'packet_seed_restore';

/**
* Seed a job's engine data from its first packet, remembering what changed.
*
* A source step's job is seeded with the first packet's engine data so a
* single-packet run has its item context. When the step fans out, each
* child starts from a copy of this parent, so withoutSeed() must be able to
* take the first packet's values back out; otherwise every child inherits
* them wherever its own packet has no value for the same key.
*
* Repeated seeding (retries) keeps the first-recorded prior values.
*
* @param array $engine Job engine data.
* @param array $seed_data Sanitized packet engine data.
* @return array Seeded engine data.
*/
public static function applySeed( array $engine, array $seed_data ): array {
$restore = is_array( $engine[ self::SEED_RESTORE_KEY ] ?? null ) ? $engine[ self::SEED_RESTORE_KEY ] : array();
$keys = is_array( $restore['keys'] ?? null ) ? $restore['keys'] : array();
$prior = is_array( $restore['prior'] ?? null ) ? $restore['prior'] : array();

foreach ( array_keys( $seed_data ) as $key ) {
$key = (string) $key;
if ( 'packet_fanout_transfer' === $key || self::SEED_RESTORE_KEY === $key || in_array( $key, $keys, true ) ) {
continue;
}
$keys[] = $key;
if ( array_key_exists( $key, $engine ) ) {
$prior[ $key ] = $engine[ $key ];
}
}

unset( $seed_data[ self::SEED_RESTORE_KEY ] );
$engine = array_replace_recursive( $engine, $seed_data );
if ( $keys ) {
$engine[ self::SEED_RESTORE_KEY ] = array(
'keys' => $keys,
'prior' => $prior,
);
}

return $engine;
}

/**
* Undo applySeed(): restore pre-seed values and drop seed-only keys.
*
* @param array $engine Engine data copied from a seeded parent.
* @return array Engine data without the first packet's item context.
*/
public static function withoutSeed( array $engine ): array {
$restore = $engine[ self::SEED_RESTORE_KEY ] ?? null;
unset( $engine[ self::SEED_RESTORE_KEY ] );
if ( ! is_array( $restore ) ) {
return $engine;
}

$prior = is_array( $restore['prior'] ?? null ) ? $restore['prior'] : array();
foreach ( (array) ( $restore['keys'] ?? array() ) as $key ) {
$key = (string) $key;
if ( array_key_exists( $key, $prior ) ) {
$engine[ $key ] = $prior[ $key ];
} else {
unset( $engine[ $key ] );
}
}

return $engine;
}

/**
* Check whether a packet-provided engine data key is reserved.
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -344,7 +344,7 @@ private static function commitReconciliationAttempt( int $job_id, array $flow_st
}

$evidence = self::claimEvidence( $flow_step_config, array_keys( $retained ), $completed, $released, $omitted, $exhausted );
$engine = array_replace_recursive( $engine, $seed_data );
$engine = PacketEngineData::applySeed( $engine, $seed_data );
if ( isset( $seed_data['packet_fanout_transfer'] ) ) {
$engine['packet_fanout_transfer'] = $seed_data['packet_fanout_transfer'];
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -415,6 +415,32 @@ public function test_process_chunk_creates_child_jobs(): void {
$this->assertEquals( 2, $parent_engine['batch_scheduled'] );
}

public function test_children_do_not_inherit_first_packets_seeded_engine_data(): void {
$parent_id = $this->create_parent_job();
$first = $this->make_data_packet( 'Placeholder' );
$second = $this->make_data_packet( 'Real Show' );
$first['metadata']['_engine_data'] = array( 'title' => 'Placeholder', 'status_flag' => 'cancelled' );
$second['metadata']['_engine_data'] = array( 'title' => 'Real Show' );

// The source step seeds the parent with the first packet's engine data.
$engine = \DataMachine\Core\PacketEngineData::applySeed( datamachine_get_engine_data( $parent_id ), $first['metadata']['_engine_data'] );
datamachine_set_engine_data( $parent_id, $engine );

$scheduler = new PipelineBatchScheduler();
$scheduler->fanOut( $parent_id, 'step_abc_123', array( $first, $second ), $engine );
$scheduler->processChunk( $parent_id );

$by_title = array();
foreach ( $this->jobs_db->get_children( $parent_id ) as $child ) {
$child_engine = datamachine_get_engine_data( (int) $child['job_id'] );
$by_title[ $child_engine['title'] ] = $child_engine;
}

$this->assertSame( 'cancelled', $by_title['Placeholder']['status_flag'] );
$this->assertArrayNotHasKey( 'status_flag', $by_title['Real Show'] );
$this->assertArrayNotHasKey( \DataMachine\Core\PacketEngineData::SEED_RESTORE_KEY, $by_title['Real Show'] );
}

public function test_scheduled_chunk_reads_persisted_batch_state_when_cache_is_stale(): void {
$parent_id = $this->create_parent_job();
$engine = $this->make_engine_snapshot( $parent_id );
Expand Down
48 changes: 48 additions & 0 deletions plugins/data-machine/tests/Unit/Core/PacketEngineDataSeedTest.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
<?php
/**
* PacketEngineData seed/restore tests.
*
* @package DataMachine\Tests\Unit\Core
*/

namespace DataMachine\Tests\Unit\Core;

use DataMachine\Core\PacketEngineData;
use WP_UnitTestCase;

class PacketEngineDataSeedTest extends WP_UnitTestCase {

public function test_seed_overlays_and_records_changed_keys(): void {
$engine = PacketEngineData::applySeed(
array( 'source_url' => 'https://parent.example', 'flow' => array( 'name' => 'F' ) ),
array( 'title' => 'First', 'status_flag' => 'cancelled', 'source_url' => 'https://item.example' )
);

$this->assertSame( 'First', $engine['title'] );
$this->assertSame( 'cancelled', $engine['status_flag'] );
$this->assertSame( 'https://item.example', $engine['source_url'] );
$this->assertSame( array( 'title', 'status_flag', 'source_url' ), $engine[ PacketEngineData::SEED_RESTORE_KEY ]['keys'] );
$this->assertSame( array( 'source_url' => 'https://parent.example' ), $engine[ PacketEngineData::SEED_RESTORE_KEY ]['prior'] );
}

public function test_without_seed_restores_prior_values_and_drops_seed_only_keys(): void {
$parent = array( 'source_url' => 'https://parent.example', 'flow' => array( 'name' => 'F' ) );
$seeded = PacketEngineData::applySeed( $parent, array( 'title' => 'First', 'status_flag' => 'cancelled', 'source_url' => 'https://item.example' ) );

$this->assertSame( $parent, PacketEngineData::withoutSeed( $seeded ) );
}

public function test_repeated_seed_keeps_first_prior_values(): void {
$parent = array( 'source_url' => 'https://parent.example' );
$once = PacketEngineData::applySeed( $parent, array( 'source_url' => 'https://a.example', 'title' => 'A' ) );
$twice = PacketEngineData::applySeed( $once, array( 'source_url' => 'https://a.example', 'title' => 'A', 'extra' => 1 ) );

$this->assertSame( $parent, PacketEngineData::withoutSeed( $twice ) );
}

public function test_without_seed_is_a_no_op_for_unseeded_engine_data(): void {
$engine = array( 'title' => 'Kept' );

$this->assertSame( $engine, PacketEngineData::withoutSeed( $engine ) );
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ public static function reference_packet_collections_in_value( mixed $value ): mi
public static function hydrate_packet_collections_with_status( mixed $value ): array { return array( 'success' => true, 'value' => $value ); }
}
class JobStatus { public static function isStatusFinal( string $status ): bool { return false; } }
class PacketEngineData {}
class PacketEngineData { public static function withoutSeed( array $engine ): array { return $engine; } }
class RunMetrics { public static function start( int $job_id, array $context ): void { unset( $job_id, $context ); } }
class EngineData {
public static array $engines = array();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ public static function isStatusFinal( string $status ): bool { return in_array(
public static function isStatusSuccess( string $status ): bool { return 'completed' === $status; }
public static function failed( string $reason ): object { return new class( $reason ) { public function __construct( private string $reason ) {} public function toString(): string { return 'failed - ' . $this->reason; } }; }
}
class PacketEngineData { public static function sanitize( array $data, int $job_id ): array { unset( $job_id ); return $data; } }
class PacketEngineData { public static function sanitize( array $data, int $job_id ): array { unset( $job_id ); return $data; } public static function applySeed( array $engine, array $seed ): array { return array_replace_recursive( $engine, $seed ); } }
class RunMetrics { public static function recordStepResult( int $job_id, string $step, array $data ): void { unset( $job_id, $step, $data ); } }
class DataPacketStore {}
}
Expand Down
Loading