diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 095390a..51d033f 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -182,6 +182,10 @@ prautoblogger/ │ │ ├── class-run-state.php # runs-table ledger + lifecycle row (ceiling/reserved/settled, pins, status) │ │ ├── class-cost-governor.php # Per-run reserve-before-call enforcement (atomic conditional UPDATE) │ │ ├── class-cost-ceiling-exception.php # Thrown on ceiling breach (run already halted) +│ │ ├── class-run-stage-state.php # Per-run per-stage state machine (idempotent resume, output snapshots) +│ │ ├── class-run-reaper.php # Stuck-run sweep + audit-payload retention (rides the #19 cron) +│ │ ├── class-audit-writer.php # run_sources / run_decisions insert layer +│ │ ├── class-pipeline-status.php # Status-transient + summary helpers (extracted from runner/worker) │ │ ├── class-logger.php # Structured logging singleton (error/warning/info/debug) │ │ ├── class-article-worker.php # Single-article generation (content + edit + publish) │ │ ├── class-pipeline-runner.php # Orchestrates pipeline; chains per-article cron jobs @@ -563,6 +567,7 @@ All prefixed with `prautoblogger_`: | `prautoblogger_target_subreddits` | JSON array of subreddits to monitor | | `prautoblogger_monthly_budget_usd` | Monthly API spend limit in USD | | `prautoblogger_per_run_cost_ceiling_usd` | v0.18.0 — per-run hard cost ceiling (USD); reserve-before-call enforced; snapshotted onto the runs row at run start; 0 = disabled. Default `PRAUTOBLOGGER_DEFAULT_RUN_CEILING_USD` ($0.50) | +| `prautoblogger_request_json_retention_days` | v0.18.0 — days before heavy audit payloads (generation_log.request_json, run_stages.meta_json) are NULLed by the daily cleanup cron; 0 = keep forever. Default `PRAUTOBLOGGER_DEFAULT_REQUEST_JSON_RETENTION_DAYS` (14) | | `prautoblogger_tone` | Content tone (informational, conversational, etc.) | | `prautoblogger_min_word_count` | Minimum article word count (default: 800) | | `prautoblogger_max_word_count` | Maximum article word count (default: 2000) | @@ -608,6 +613,7 @@ Stored on every PRAutoBlogger-generated post: | `_prautoblogger_editor_notes` | Chief editor's review notes | | `_prautoblogger_generated_at` | ISO 8601 timestamp of generation | | `_prautoblogger_research_sources` | JSON array of source URLs used | +| `_prautoblogger_idea_hash` | v0.18.0 — stable hash of the idea (title|topic); with `_prautoblogger_run_id` forms the post-creation idempotency key | | `_prautoblogger_og_image_id` | Attachment ID of the composed 1200×630 OG variant (v0.17.0; the rebuilt SEO stage emits `og:image` from this — key name frozen) | | `_prautoblogger_square_image_id` | Attachment ID of the composed 1080×1080 square variant (v0.17.0; stored now, no consumer yet) | diff --git a/CHANGELOG.md b/CHANGELOG.md index e559c35..aea9b1b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,6 +56,30 @@ behavior change** to the Economy (single-pass) and multi-step publish paths. 0 = disabled). Calls outside a run, the monthly budget, and the Cloudflare AI Gateway path behave exactly as before. +- **Run state machine + idempotent resume** (`PRAutoBlogger_Run_Stage_State`): per-run + per-stage rows keyed `run_id + stage + agent_role + item_key` (item_key scopes + article-level stages — one run_id spans all N articles of a batch; agent_role is the + Phase-2 quorum dimension, schema-only for now). Done stages snapshot their output and are + REUSED on re-entry — never re-run, never re-charged (writer stages via + `Content_Generator::set_run_item()`, the editorial review via its stored verdict, a fully + published item is skipped outright). **Post creation is keyed by `_prautoblogger_run_id` + + new `_prautoblogger_idea_hash` meta (check-before-insert)** so a retried run cannot + create a duplicate post. Run-level research/analysis checkpoints ride Pipeline_Runner; + completion/failure marks land on the runs row (Executor catch paths, orphan aborts, + monthly-budget aborts). +- **Reaper extension** (`PRAutoBlogger_Run_Reaper`, same daily cron as the #19 research + reaper — no new schedule): stages stuck `running` > 2× expected stage wall-clock and runs + silent > 2× expected run wall-clock are marked `failed` (filterable expectations); their + open stages are reaped; such runs are queryable as **"incomplete"** via + `Run_Reaper::incomplete_runs()`. Also enforces **retention**: `request_json` on the + generation log and stage output snapshots are NULLed after R days — R from the new + **Audit Payload Retention** setting (`prautoblogger_request_json_retention_days`, default + `PRAUTOBLOGGER_DEFAULT_REQUEST_JSON_RETENTION_DAYS` = 14, 0 = keep forever). +- **Audit writes** (`PRAutoBlogger_Audit_Writer`): every reviewed article records its + editorial verdict in `run_decisions`; every published article records the sources that + fed it in `run_sources` (the source_ids_json / `_prautoblogger_research_sources` + consolidation for new runs; no backfill). + ### Changed - `Content_Generator`'s four stage methods now share one `execute_stage()` helper (the Opik span + dispatch + cost-log boilerplate was identical); file drops from 301 to ~220 lines. @@ -64,6 +88,9 @@ behavior change** to the Economy (single-pass) and multi-step publish paths. reporting query); the Cost_Tracker method remains as a delegating wrapper. - `OpenRouter_Provider`'s API-key fetch + format validation moved to `OpenRouter_Request_Builder::resolve_api_key()` (300-line cap); identical messages/behavior. +- Pipeline_Runner's and Article_Worker's duplicated status-transient/summary helpers moved + verbatim to the new `PRAutoBlogger_Pipeline_Status` (Pipeline_Runner back under the + 300-line cap at 271). ### Fixed - ARCHITECTURE.md documented the multi-step edit stage as `edit`; the code has always diff --git a/CONVENTIONS.md b/CONVENTIONS.md index 7c5243e..4382ecf 100644 --- a/CONVENTIONS.md +++ b/CONVENTIONS.md @@ -204,13 +204,18 @@ the uninstall prefix-sweep effective), then cover it in the unit ladder tests. ## How To: Add a New Pipeline Stage -The writing pipeline is an ordered array of stages. Each stage is a class method on `Content_Generator`. +Writing stages are methods on `Content_Generator` that delegate to its `execute_stage()` +helper (Opik span + LLM dispatch + cost log + run-stage state in one place). -1. Add the stage method to `class-content-generator.php`: `protected function stage_{name}(PRAutoBlogger_Content_Request $request): string` -2. Register the stage in the `get_pipeline_stages()` method for the appropriate mode (single_pass or multi_step) -3. Each stage receives the accumulated context and returns its output -4. Each stage logs its cost via `Cost_Tracker::log_api_call()` -5. If a stage fails after retries, the entire pipeline fails (no partial publishes) +1. Add the stage method to `class-content-generator.php` and call + `$this->execute_stage( $stage, $span_name, $user_prompt, $request, $model, $options )` +2. Wire it into `generate_multi_step()` (or the relevant mode path) +3. Add the stage to `PRAutoBlogger_Stage_Display_Map` (label, default agent role, prompt key) + and — if it has registry-managed copy — a prompt key in the defaults classes +4. Cost logging and per-run stage state (idempotent resume, output snapshot) come free from + `execute_stage()`; the per-run cost governor guards the call inside the provider +5. If a stage fails after retries, the entire pipeline fails (no partial publishes); the + stage row is failed by the worker's catch-all and the run is swept by Run_Reaper --- diff --git a/includes/admin/class-settings-fields-extended.php b/includes/admin/class-settings-fields-extended.php index ab0e5c4..8e53f38 100644 --- a/includes/admin/class-settings-fields-extended.php +++ b/includes/admin/class-settings-fields-extended.php @@ -113,6 +113,16 @@ class PRAutoBlogger_Settings_Fields_Extended { 'step' => '0.01', 'description' => __( 'Hard stop when reached. 0 = unlimited.', 'prautoblogger' ), ), + array( + 'id' => 'prautoblogger_request_json_retention_days', + 'label' => __( 'Audit Payload Retention (days)', 'prautoblogger' ), + 'type' => 'number', + 'section' => 'prautoblogger_schedule', + 'default' => PRAUTOBLOGGER_DEFAULT_REQUEST_JSON_RETENTION_DAYS, + 'min' => 0, + 'step' => '1', + 'description' => __( 'Heavy audit payloads (request_json on the generation log, stage output snapshots) are emptied after this many days by the daily cleanup cron. Cost/token rows are kept forever. 0 = keep payloads forever.', 'prautoblogger' ), + ), array( 'id' => 'prautoblogger_per_run_cost_ceiling_usd', 'label' => __( 'Per-Run Cost Ceiling (USD)', 'prautoblogger' ), diff --git a/includes/class-activator.php b/includes/class-activator.php index f42cb91..7892977 100644 --- a/includes/class-activator.php +++ b/includes/class-activator.php @@ -69,6 +69,7 @@ class PRAutoBlogger_Activator { 'prautoblogger_target_subreddits' => '[]', 'prautoblogger_monthly_budget_usd' => 50.00, 'prautoblogger_per_run_cost_ceiling_usd' => PRAUTOBLOGGER_DEFAULT_RUN_CEILING_USD, + 'prautoblogger_request_json_retention_days' => PRAUTOBLOGGER_DEFAULT_REQUEST_JSON_RETENTION_DAYS, 'prautoblogger_tone' => 'informational', 'prautoblogger_min_word_count' => 800, 'prautoblogger_max_word_count' => 2000, diff --git a/includes/class-executor.php b/includes/class-executor.php index a2edfc2..f3f0fa8 100644 --- a/includes/class-executor.php +++ b/includes/class-executor.php @@ -53,10 +53,25 @@ class PRAutoBlogger_Executor { sprintf( 'Daily generation %s: %s', get_class( $e ), $e->getMessage() ), 'scheduler' ); + $this->mark_current_run_failed(); PRAutoBlogger_Generation_Lock::release(); } } + /** + * Mark this process's run failed after a fatal pipeline error + * (no-op outside a run; a governor-halted run stays halted — final + * states are sticky). + * + * @return void + */ + private function mark_current_run_failed(): void { + $run_id = PRAutoBlogger_Run_Context::current_run_id(); + if ( null !== $run_id ) { + PRAutoBlogger_Run_State::mark_status( $run_id, 'failed' ); + } + } + /** * Handle the metrics collection cron event. * Side effects: API calls to GA4, database writes to ab_content_scores. @@ -209,6 +224,7 @@ class PRAutoBlogger_Executor { ), self::STATUS_TTL ); + $this->mark_current_run_failed(); PRAutoBlogger_Generation_Lock::release(); } } @@ -286,6 +302,12 @@ class PRAutoBlogger_Executor { /** Clean up an orphaned generation run and report error. */ private function abort_orphaned_run( string $message ): void { + // v0.18.0: the dead run renders as failed/"incomplete" in audit + // queries instead of silently lingering as 'running'. + $queue = get_option( 'prautoblogger_article_queue' ); + if ( is_array( $queue ) && ! empty( $queue['run_id'] ) ) { + PRAutoBlogger_Run_State::mark_status( (string) $queue['run_id'], 'failed' ); + } delete_transient( self::STATUS_TRANSIENT ); delete_option( 'prautoblogger_article_queue' ); PRAutoBlogger_Generation_Lock::release(); diff --git a/includes/class-prautoblogger.php b/includes/class-prautoblogger.php index 24c5b1a..ee4c417 100644 --- a/includes/class-prautoblogger.php +++ b/includes/class-prautoblogger.php @@ -150,6 +150,10 @@ class PRAutoBlogger { // pipeline dies before amortize_research_costs runs). add_action( 'prautoblogger_reap_orphan_research_rows', array( 'PRAutoBlogger_Research_Reaper', 'on_cron' ) ); + // v0.18.0: stuck-run sweep + audit-payload retention rides the SAME + // existing daily cron (no new schedule). See class-run-reaper.php. + add_action( 'prautoblogger_reap_orphan_research_rows', array( 'PRAutoBlogger_Run_Reaper', 'on_cron' ) ); + // v0.8.1: WP-CLI manual trigger for the reaper. Only registers when // WP-CLI is present; no-op in normal HTTP requests. if ( defined( 'WP_CLI' ) && WP_CLI ) { diff --git a/includes/core/class-article-worker.php b/includes/core/class-article-worker.php index 3cae065..7d707be 100644 --- a/includes/core/class-article-worker.php +++ b/includes/core/class-article-worker.php @@ -9,8 +9,15 @@ declare(strict_types=1); * Extracted from Pipeline_Runner so each chained cron job has a focused, * single-responsibility worker. One worker = one article = one PHP process. * + * v0.18.0: per-article stage state (run_stages, item-scoped) makes resume + * idempotent — an already-published item is skipped outright, a done + * review is reused from its snapshot, and the writer reuses done LLM + * stages via Content_Generator::set_run_item(). Editorial verdicts and + * the sources that fed the article are persisted to the audit tables. + * * Triggered by: Pipeline_Runner (directly for article 1, via cron for 2..N). - * Dependencies: Content_Generator, Chief_Editor, Publisher, Cost_Tracker. + * Dependencies: Content_Generator, Chief_Editor, Publisher, Cost_Tracker, + * Run_Stage_State, Audit_Writer. * * Opik instrumentation: Initializes trace context at start, creates spans * for content generation and editorial review. Feature-flag gated (default off). @@ -48,6 +55,27 @@ class PRAutoBlogger_Article_Worker { * @return array{generated: int, published: int, rejected: int, cost: float} */ public function generate( PRAutoBlogger_Article_Idea $idea ): array { + $run_id = (string) ( $this->cost_tracker->get_run_id() ?? '' ); + $item = PRAutoBlogger_Run_Stage_State::item_key_for_idea( $idea ); + + $result = array( + 'generated' => 0, + 'published' => 0, + 'rejected' => 0, + 'cost' => 0.0, + ); + + // Idempotent resume: this item already completed its publish stage + // in this run — re-dispatching it must not create a second post or + // charge a second cent. + if ( '' !== $run_id && PRAutoBlogger_Run_Stage_State::is_done( $run_id, 'publish', '', $item ) ) { + PRAutoBlogger_Logger::instance()->info( + sprintf( 'Skipping "%s" — already completed in run %s (idempotent resume).', $idea->get_topic(), $run_id ), + 'pipeline' + ); + return $result; + } + // Initialize Opik trace context for this article generation. $opik = $this->should_trace_with_opik(); if ( $opik ) { @@ -56,6 +84,7 @@ class PRAutoBlogger_Article_Worker { $llm = new PRAutoBlogger_OpenRouter_Provider(); $generator = new PRAutoBlogger_Content_Generator( $llm, $this->cost_tracker ); + $generator->set_run_item( '' !== $run_id ? $run_id : null, $item ); $editor = new PRAutoBlogger_Chief_Editor( $llm, $this->cost_tracker ); $publisher = new PRAutoBlogger_Publisher(); @@ -65,22 +94,16 @@ class PRAutoBlogger_Article_Worker { true ); - $result = array( - 'generated' => 0, - 'published' => 0, - 'rejected' => 0, - 'cost' => 0.0, - ); - try { - $this->broadcast_stage( __( 'Generating article draft via AI…', 'prautoblogger' ) ); + PRAutoBlogger_Pipeline_Status::broadcast( __( 'Generating article draft via AI…', 'prautoblogger' ) ); $content = $generator->generate( $idea ); $result['generated'] = 1; - $this->broadcast_stage( __( 'Running editorial pass…', 'prautoblogger' ) ); - $review = $editor->review( $content, $idea ); + PRAutoBlogger_Pipeline_Status::broadcast( __( 'Running editorial pass…', 'prautoblogger' ) ); + $review = $this->review_with_state( $editor, $content, $idea, $run_id, $item ); - $this->broadcast_stage( __( 'Saving and publishing…', 'prautoblogger' ) ); + PRAutoBlogger_Pipeline_Status::broadcast( __( 'Saving and publishing…', 'prautoblogger' ) ); + PRAutoBlogger_Run_Stage_State::start( $run_id, 'publish', '', $item ); $this->publish_or_draft( $content, $idea, @@ -89,7 +112,13 @@ class PRAutoBlogger_Article_Worker { $auto_publish, $result ); + PRAutoBlogger_Run_Stage_State::done( $run_id, 'publish', '', $item ); + + PRAutoBlogger_Audit_Writer::record_idea_sources( $run_id, $idea ); } catch ( \Throwable $e ) { + if ( '' !== $run_id ) { + PRAutoBlogger_Run_Stage_State::fail_open_for_item( $run_id, $item ); + } PRAutoBlogger_Logger::instance()->error( sprintf( 'Article generation %s for "%s": %s', @@ -111,6 +140,63 @@ class PRAutoBlogger_Article_Worker { return $result; } + /** + * Editorial review with stage state: a done review is reused from its + * snapshot (never re-charged); a fresh review is checkpointed and its + * verdict recorded to the run_decisions audit table. + * + * Side effects: run_stages upserts, one run_decisions insert, and — + * when not resuming — one LLM call via Chief_Editor. + * + * @param PRAutoBlogger_Chief_Editor $editor Editor agent. + * @param string $content Generated HTML content. + * @param PRAutoBlogger_Article_Idea $idea The idea under review. + * @param string $run_id Run UUID ('' outside a run). + * @param string $item Stage item key for this idea. + * @return PRAutoBlogger_Editorial_Review + */ + private function review_with_state( + PRAutoBlogger_Chief_Editor $editor, + string $content, + PRAutoBlogger_Article_Idea $idea, + string $run_id, + string $item + ): PRAutoBlogger_Editorial_Review { + if ( '' !== $run_id ) { + $snapshot = PRAutoBlogger_Run_Stage_State::get_output( $run_id, 'review', '', $item ); + if ( null !== $snapshot ) { + $data = json_decode( $snapshot, true ); + if ( is_array( $data ) && isset( $data['verdict'] ) ) { + PRAutoBlogger_Logger::instance()->info( + sprintf( 'Reusing completed review for "%s" (idempotent resume).', $idea->get_topic() ), + 'pipeline' + ); + return new PRAutoBlogger_Editorial_Review( $data ); + } + } + PRAutoBlogger_Run_Stage_State::start( $run_id, 'review', '', $item ); + } + + $review = $editor->review( $content, $idea ); + + if ( '' !== $run_id ) { + $snapshot = (string) wp_json_encode( + array( + 'verdict' => $review->get_verdict(), + 'notes' => $review->get_notes(), + 'revised_content' => $review->get_revised_content(), + 'quality_score' => $review->get_quality_score(), + 'seo_score' => $review->get_seo_score(), + 'issues' => $review->get_issues(), + ) + ); + PRAutoBlogger_Run_Stage_State::done( $run_id, 'review', '', $item, $snapshot ); + PRAutoBlogger_Audit_Writer::record_decision( $run_id, 'review', $review->get_verdict(), $review->get_notes() ); + } + + return $review; + } + /** * Publish approved content or save as draft. * @@ -153,21 +239,6 @@ class PRAutoBlogger_Article_Worker { } } - /** - * Update the generation status transient with the current stage. - * - * @param string $stage Human-readable stage description. - */ - private function broadcast_stage( string $stage ): void { - $transient_key = 'prautoblogger_generation_status'; - $current = get_transient( $transient_key ); - if ( is_array( $current ) && 'running' === ( $current['status'] ?? '' ) ) { - $current['stage'] = $stage; - $current['last_updated'] = time(); - set_transient( $transient_key, $current, 600 ); - } - } - /** * Check if Opik tracing is enabled and has credentials. * diff --git a/includes/core/class-audit-writer.php b/includes/core/class-audit-writer.php new file mode 100644 index 0000000..55537b5 --- /dev/null +++ b/includes/core/class-audit-writer.php @@ -0,0 +1,132 @@ +prefix . 'prautoblogger_run_decisions'; + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching + $found = $wpdb->get_var( $wpdb->prepare( 'SHOW TABLES LIKE %s', $table ) ); + self::$tables_ok = ( $found === $table ); + return self::$tables_ok; + } + + /** + * Persist a stage decision for a run. + * + * @param string $run_id Run UUID. + * @param string $stage Deciding stage (see Stage_Display_Map). + * @param string $verdict e.g. 'approved', 'revised', 'rejected'. + * @param string $rationale Why (editor notes, governor message, …). + * @param float|null $citation_score Phase-2 citation score (null in Phase 1). + * @return void + */ + public static function record_decision( string $run_id, string $stage, string $verdict, string $rationale = '', ?float $citation_score = null ): void { + if ( '' === $run_id || ! self::is_available() ) { + return; + } + global $wpdb; + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $wpdb->insert( + $wpdb->prefix . 'prautoblogger_run_decisions', + array( + 'run_id' => $run_id, + 'stage' => $stage, + 'verdict' => $verdict, + 'rationale' => $rationale, + 'citation_score' => $citation_score, + 'created_at' => current_time( 'mysql' ), + ) + ); + } + + /** + * Record the collected sources that fed a generated article (kept = 1; + * Phase-2 curation adds discard rows with reasons + quality scores). + * + * Resolves the idea's source_data ids to permalinks in one query. + * + * @param string $run_id Run UUID. + * @param PRAutoBlogger_Article_Idea $idea The generated idea. + * @return void + */ + public static function record_idea_sources( string $run_id, PRAutoBlogger_Article_Idea $idea ): void { + if ( '' === $run_id || ! self::is_available() ) { + return; + } + $source_ids = array_filter( array_map( 'intval', $idea->get_source_ids() ) ); + if ( empty( $source_ids ) ) { + return; + } + + global $wpdb; + $source_table = $wpdb->prefix . 'prautoblogger_source_data'; + $placeholders = implode( ',', array_fill( 0, count( $source_ids ), '%d' ) ); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.NotPrepared + $rows = $wpdb->get_results( + // phpcs:ignore WordPress.DB.PreparedSQL.NotPrepared -- $placeholders is a fixed %d list. + $wpdb->prepare( "SELECT id, source_type, permalink FROM {$source_table} WHERE id IN ({$placeholders})", $source_ids ), + ARRAY_A + ); + if ( ! is_array( $rows ) ) { + return; + } + + foreach ( $rows as $row ) { + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $wpdb->insert( + $wpdb->prefix . 'prautoblogger_run_sources', + array( + 'run_id' => $run_id, + 'agent_role' => 'analyst', + 'source_url' => (string) ( $row['permalink'] ?? '' ), + 'doi' => null, + 'kept' => 1, + 'reason' => sprintf( 'Cited by analyzer for "%s" (%s #%d)', $idea->get_topic(), (string) $row['source_type'], (int) $row['id'] ), + 'quality_score' => null, + 'created_at' => current_time( 'mysql' ), + ) + ); + } + } + + /** Reset per-request caches (tests). */ + public static function flush_cache(): void { + self::$tables_ok = null; + } +} diff --git a/includes/core/class-content-generator.php b/includes/core/class-content-generator.php index 3e77c28..a2c8d80 100644 --- a/includes/core/class-content-generator.php +++ b/includes/core/class-content-generator.php @@ -19,6 +19,12 @@ class PRAutoBlogger_Content_Generator { private PRAutoBlogger_LLM_Provider_Interface $llm; private PRAutoBlogger_Cost_Tracker $cost_tracker; + /** @var string|null Run UUID for stage-state checkpointing (null = off). */ + private ?string $run_id = null; + + /** @var string|null Stage item key scoping this article within the run. */ + private ?string $item_key = null; + public function __construct( PRAutoBlogger_LLM_Provider_Interface $llm, PRAutoBlogger_Cost_Tracker $cost_tracker @@ -27,6 +33,25 @@ class PRAutoBlogger_Content_Generator { $this->cost_tracker = $cost_tracker; } + /** + * Enable per-stage state checkpointing for one article item (v0.18.0). + * + * When set, every LLM stage (outline/draft/polish or the single-pass + * draft) records pending->running->done state and snapshots its output; + * a re-entered done stage returns the snapshot WITHOUT a new LLM call + * (never re-charged). When unset (default), behavior is exactly + * pre-v0.18.0 — used by callers outside a run (e.g. eval mode). + * + * @param string|null $run_id Run UUID, or null to disable. + * @param string|null $item_key Item key from Run_Stage_State::item_key_for_idea(). + * @return $this + */ + public function set_run_item( ?string $run_id, ?string $item_key ): self { + $this->run_id = $run_id; + $this->item_key = $item_key; + return $this; + } + /** * Generate a blog post from an article idea. * @@ -159,6 +184,19 @@ class PRAutoBlogger_Content_Generator { array $options, ?string $prompt_key = null ): string { + // Idempotent resume: a done stage returns its snapshot, no LLM call. + if ( null !== $this->run_id && null !== $this->item_key ) { + $cached = PRAutoBlogger_Run_Stage_State::get_output( $this->run_id, $stage, '', $this->item_key ); + if ( null !== $cached ) { + PRAutoBlogger_Logger::instance()->info( + sprintf( 'Reusing completed %s stage output (idempotent resume).', $stage ), + 'pipeline' + ); + return $cached; + } + PRAutoBlogger_Run_Stage_State::start( $this->run_id, $stage, '', $this->item_key ); + } + $ctx = PRAutoBlogger_Opik_Trace_Context::current(); $span_id = $ctx->start_span( @@ -209,6 +247,17 @@ class PRAutoBlogger_Content_Generator { $prompt_key ); + if ( null !== $this->run_id && null !== $this->item_key ) { + PRAutoBlogger_Run_Stage_State::done( + $this->run_id, + $stage, + '', + $this->item_key, + (string) $response['content'], + $this->llm->estimate_cost( $response['model'], (int) $response['prompt_tokens'], (int) $response['completion_tokens'] ) + ); + } + return $response['content']; } diff --git a/includes/core/class-pipeline-runner.php b/includes/core/class-pipeline-runner.php index 736a4b8..f34f86d 100644 --- a/includes/core/class-pipeline-runner.php +++ b/includes/core/class-pipeline-runner.php @@ -65,6 +65,7 @@ class PRAutoBlogger_Pipeline_Runner { $ideas = $this->orchestrate( $cost_tracker ); if ( empty( $ideas ) ) { + PRAutoBlogger_Run_State::mark_status( (string) $cost_tracker->get_run_id(), 'done' ); return array( 'generated' => 0, 'published' => 0, @@ -84,7 +85,8 @@ class PRAutoBlogger_Pipeline_Runner { } else { // Single-article run — amortize research costs now (1 article = full cost). PRAutoBlogger_Post_Assembler::amortize_research_costs( $cost_tracker->get_run_id() ); - $this->log_summary( $result ); + PRAutoBlogger_Pipeline_Status::log_summary( $result ); + PRAutoBlogger_Run_State::mark_status( (string) $cost_tracker->get_run_id(), 'done' ); } return $result; @@ -115,6 +117,7 @@ class PRAutoBlogger_Pipeline_Runner { 'Budget limit reached. Aborting remaining queued articles.', 'pipeline' ); + PRAutoBlogger_Run_State::mark_status( (string) $queue['run_id'], 'failed' ); $this->finish_queue(); return; } @@ -152,15 +155,16 @@ class PRAutoBlogger_Pipeline_Runner { if ( ! empty( $queue['ideas'] ) ) { update_option( self::QUEUE_KEY, $queue, false ); - $this->update_queue_status( $queue ); + PRAutoBlogger_Pipeline_Status::update_queue_progress( $queue ); $this->schedule_next(); } else { // All articles done — amortize shared research costs across them. PRAutoBlogger_Post_Assembler::amortize_research_costs( $queue['run_id'] ); - $this->write_final_status( $queue['results'] ); + PRAutoBlogger_Pipeline_Status::write_final( $queue['results'] ); delete_option( self::QUEUE_KEY ); PRAutoBlogger_Generation_Lock::release(); - $this->log_summary( $queue['results'] ); + PRAutoBlogger_Pipeline_Status::log_summary( $queue['results'] ); + PRAutoBlogger_Run_State::mark_status( (string) $queue['run_id'], 'done' ); } } @@ -175,20 +179,26 @@ class PRAutoBlogger_Pipeline_Runner { private function orchestrate( PRAutoBlogger_Cost_Tracker $cost_tracker ): array { $target = absint( get_option( 'prautoblogger_daily_article_target', 1 ) ); + $run_id = (string) $cost_tracker->get_run_id(); + $enabled = json_decode( get_option( 'prautoblogger_enabled_sources', '["reddit"]' ), true ); $source_labels = is_array( $enabled ) ? implode( ', ', $enabled ) : 'reddit'; /* translators: %s is a comma-separated list of enabled source names, e.g. "reddit, llm_research". */ - $this->broadcast_stage( sprintf( __( 'Collecting sources from %s…', 'prautoblogger' ), $source_labels ) ); + PRAutoBlogger_Pipeline_Status::broadcast( sprintf( __( 'Collecting sources from %s…', 'prautoblogger' ), $source_labels ) ); + PRAutoBlogger_Run_Stage_State::start( $run_id, 'research' ); ( new PRAutoBlogger_Source_Collector() ) ->set_cost_tracker( $cost_tracker ) ->collect_from_all_sources(); + PRAutoBlogger_Run_Stage_State::done( $run_id, 'research' ); - $this->broadcast_stage( __( 'Analyzing topics and scoring…', 'prautoblogger' ) ); + PRAutoBlogger_Pipeline_Status::broadcast( __( 'Analyzing topics and scoring…', 'prautoblogger' ) ); + PRAutoBlogger_Run_Stage_State::start( $run_id, 'analysis' ); $llm = new PRAutoBlogger_OpenRouter_Provider(); $analyzer = new PRAutoBlogger_Content_Analyzer( $llm, $cost_tracker ); $analysis = $analyzer->analyze_recent_data( max( $target * 2, 6 ) ); + PRAutoBlogger_Run_Stage_State::done( $run_id, 'analysis' ); - $this->broadcast_stage( __( 'Selecting best topic…', 'prautoblogger' ) ); + PRAutoBlogger_Pipeline_Status::broadcast( __( 'Selecting best topic…', 'prautoblogger' ) ); $scorer = new PRAutoBlogger_Idea_Scorer(); $scorer->set_skip_dedup( $this->skip_dedup ); $ideas = $scorer->score_and_rank( $analysis, $target ); @@ -223,7 +233,7 @@ class PRAutoBlogger_Pipeline_Runner { ); update_option( self::QUEUE_KEY, $queue, false ); - $this->update_queue_status( $queue ); + PRAutoBlogger_Pipeline_Status::update_queue_progress( $queue ); $this->schedule_next(); PRAutoBlogger_Logger::instance()->info( @@ -248,71 +258,13 @@ class PRAutoBlogger_Pipeline_Runner { ); } - /** Update the frontend status transient with queue progress. */ - private function update_queue_status( array $queue ): void { - $done = $queue['results']['generated'] + $queue['results']['rejected']; - $total = $done + count( $queue['ideas'] ); - - $current = get_transient( 'prautoblogger_generation_status' ); - set_transient( - 'prautoblogger_generation_status', - array( - 'status' => 'running', - 'stage' => sprintf( __( 'Generating article %1$d of %2$d…', 'prautoblogger' ), $done + 1, $total ), - 'started' => is_array( $current ) ? ( $current['started'] ?? time() ) : time(), - 'last_updated' => time(), - ), - 600 - ); - } - - /** Write the final "complete" status transient. */ - private function write_final_status( array $r ): void { - set_transient( - 'prautoblogger_generation_status', - array( - 'status' => 'complete', - 'generated' => $r['generated'], - 'published' => $r['published'], - 'rejected' => $r['rejected'], - 'cost' => $r['cost'], - ), - 600 - ); - } - /** Clean up a stale or empty queue. */ private function finish_queue(): void { $queue = get_option( self::QUEUE_KEY ); if ( is_array( $queue ) && isset( $queue['results'] ) ) { - $this->write_final_status( $queue['results'] ); + PRAutoBlogger_Pipeline_Status::write_final( $queue['results'] ); } delete_option( self::QUEUE_KEY ); PRAutoBlogger_Generation_Lock::release(); } - - /** Log the pipeline summary. */ - private function log_summary( array $r ): void { - PRAutoBlogger_Logger::instance()->info( - sprintf( - 'Pipeline complete: %d generated, %d published, %d rejected. Cost: $%.4f', - $r['generated'], - $r['published'], - $r['rejected'], - $r['cost'] - ), - 'pipeline' - ); - } - - /** Update the generation status transient with the current stage. */ - private function broadcast_stage( string $stage ): void { - $key = 'prautoblogger_generation_status'; - $current = get_transient( $key ); - if ( is_array( $current ) && 'running' === ( $current['status'] ?? '' ) ) { - $current['stage'] = $stage; - $current['last_updated'] = time(); - set_transient( $key, $current, 600 ); - } - } } diff --git a/includes/core/class-pipeline-status.php b/includes/core/class-pipeline-status.php new file mode 100644 index 0000000..6e7d01a --- /dev/null +++ b/includes/core/class-pipeline-status.php @@ -0,0 +1,106 @@ + 'running', + /* translators: 1: current article number, 2: total articles. */ + 'stage' => sprintf( __( 'Generating article %1$d of %2$d…', 'prautoblogger' ), $done + 1, $total ), + 'started' => is_array( $current ) ? ( $current['started'] ?? time() ) : time(), + 'last_updated' => time(), + ), + self::TTL + ); + } + + /** + * Write the final "complete" status. + * + * @param array $r Result counters (generated/published/rejected/cost). + * @return void + */ + public static function write_final( array $r ): void { + set_transient( + self::KEY, + array( + 'status' => 'complete', + 'generated' => $r['generated'], + 'published' => $r['published'], + 'rejected' => $r['rejected'], + 'cost' => $r['cost'], + ), + self::TTL + ); + } + + /** + * Log the pipeline summary line. + * + * @param array $r Result counters (generated/published/rejected/cost). + * @return void + */ + public static function log_summary( array $r ): void { + PRAutoBlogger_Logger::instance()->info( + sprintf( + 'Pipeline complete: %d generated, %d published, %d rejected. Cost: $%.4f', + $r['generated'], + $r['published'], + $r['rejected'], + $r['cost'] + ), + 'pipeline' + ); + } +} diff --git a/includes/core/class-publisher.php b/includes/core/class-publisher.php old mode 100755 new mode 100644 index be46fe6..d40b5cd --- a/includes/core/class-publisher.php +++ b/includes/core/class-publisher.php @@ -82,6 +82,22 @@ class PRAutoBlogger_Publisher { ?string $run_id = null, ?PRAutoBlogger_Cost_Tracker $cost_tracker = null ): int { + // v0.18.0 idempotency: post creation is keyed by run + idea + // (check-before-insert) so a retried/resumed run cannot create a + // duplicate post. One run_id spans all N articles of a batch, so + // the idea hash disambiguates within the run. + $idea_hash = PRAutoBlogger_Run_Stage_State::idea_hash( $idea ); + if ( null !== $run_id && '' !== $run_id ) { + $existing_id = $this->find_existing_post( $run_id, $idea_hash ); + if ( $existing_id > 0 ) { + PRAutoBlogger_Logger::instance()->info( + sprintf( 'Post for run %s / idea %s already exists (ID=%d) — skipping duplicate insert.', $run_id, $idea_hash, $existing_id ), + 'publisher' + ); + return $existing_id; + } + } + // Clean LLM artifacts, then inject peptide hyperlinks deterministically // before the content enters WordPress. Peptide linker no-ops gracefully // if PR Core is not active. @@ -94,7 +110,7 @@ class PRAutoBlogger_Publisher { 'post_status' => $post_status, 'post_type' => 'post', 'post_author' => PRAutoBlogger_Post_Assembler::get_default_author_id(), - 'meta_input' => $this->build_meta( $idea, $review, $run_id ), + 'meta_input' => $this->build_meta( $idea, $review, $run_id, $idea_hash ), ); /** @see class-prautoblogger.php — listeners registered in main loader. */ @@ -140,7 +156,8 @@ class PRAutoBlogger_Publisher { private function build_meta( PRAutoBlogger_Article_Idea $idea, PRAutoBlogger_Editorial_Review $review, - ?string $run_id = null + ?string $run_id = null, + string $idea_hash = '' ): array { $meta = array( '_prautoblogger_generated' => '1', @@ -160,6 +177,42 @@ class PRAutoBlogger_Publisher { if ( null !== $run_id && '' !== $run_id ) { $meta['_prautoblogger_run_id'] = $run_id; } + if ( '' !== $idea_hash ) { + $meta['_prautoblogger_idea_hash'] = $idea_hash; + } return $meta; } + + /** + * Find a post already created for this (run, idea) pair. + * + * Direct postmeta self-join (any post_status, including trash) — the + * idempotency check must see drafts and pending posts too. + * + * @param string $run_id Pipeline run UUID. + * @param string $idea_hash Idea hash from Run_Stage_State::idea_hash(). + * @return int Existing post ID, or 0 when none. + */ + private function find_existing_post( string $run_id, string $idea_hash ): int { + global $wpdb; + if ( null === $wpdb ) { + return 0; + } + $postmeta = $wpdb->prefix . 'postmeta'; + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching + $post_id = $wpdb->get_var( + $wpdb->prepare( + "SELECT pm_run.post_id FROM {$postmeta} pm_run + INNER JOIN {$postmeta} pm_hash + ON pm_hash.post_id = pm_run.post_id + AND pm_hash.meta_key = '_prautoblogger_idea_hash' + AND pm_hash.meta_value = %s + WHERE pm_run.meta_key = '_prautoblogger_run_id' AND pm_run.meta_value = %s + LIMIT 1", + $idea_hash, + $run_id + ) + ); + return null !== $post_id ? (int) $post_id : 0; + } } diff --git a/includes/core/class-run-reaper.php b/includes/core/class-run-reaper.php new file mode 100644 index 0000000..6f1c86e --- /dev/null +++ b/includes/core/class-run-reaper.php @@ -0,0 +1,270 @@ + 0, + 'runs_failed' => 0, + 'payloads_pruned' => 0, + ); + + try { + $stats['stages_failed'] = self::sweep_stuck_stages(); + $stats['runs_failed'] = self::sweep_stuck_runs(); + $stats['payloads_pruned'] = self::prune_payloads(); + } catch ( \Throwable $e ) { + PRAutoBlogger_Logger::instance()->warning( + 'Run reaper encountered an error: ' . $e->getMessage(), + 'run-reaper' + ); + } + + return $stats; + } + + /** + * Runs that died mid-pipeline, for audit surfaces: failed/halted runs + * that still have non-done stage rows ("incomplete"). + * + * @param int $limit Max rows (default 20). + * @return array> + */ + public static function incomplete_runs( int $limit = 20 ): array { + if ( ! PRAutoBlogger_Run_State::is_available() || ! PRAutoBlogger_Run_Stage_State::is_available() ) { + return array(); + } + global $wpdb; + $runs = PRAutoBlogger_Run_State::table_name(); + $stages = PRAutoBlogger_Run_Stage_State::table_name(); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching + $rows = $wpdb->get_results( + $wpdb->prepare( + "SELECT r.* FROM {$runs} r + WHERE r.status IN ('failed','halted') + AND EXISTS (SELECT 1 FROM {$stages} s WHERE s.run_id = r.run_id AND s.status != 'done') + ORDER BY r.updated_at DESC LIMIT %d", + $limit + ), + ARRAY_A + ); + return is_array( $rows ) ? $rows : array(); + } + + /** + * Stage sweep: `running` rows older than 2× expected stage wall-clock + * become `failed`. + * + * @return int Rows swept. + */ + private static function sweep_stuck_stages(): int { + if ( ! PRAutoBlogger_Run_Stage_State::is_available() ) { + return 0; + } + global $wpdb; + $table = PRAutoBlogger_Run_Stage_State::table_name(); + + /** + * Filter the expected per-stage wall-clock seconds used by the + * stuck-stage sweep (threshold is 2× this value). + * + * @param int $seconds Expected seconds (default 300). + */ + $expected = (int) apply_filters( 'prautoblogger_filter_stage_expected_seconds', self::EXPECTED_STAGE_SECONDS ); + $cutoff = gmdate( 'Y-m-d H:i:s', time() - ( 2 * $expected ) ); + $now = current_time( 'mysql' ); + + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $swept = $wpdb->query( + $wpdb->prepare( + "UPDATE {$table} SET status = 'failed', updated_at = %s, finished_at = %s + WHERE status = 'running' AND updated_at < %s", + $now, + $now, + $cutoff + ) + ); + return is_numeric( $swept ) ? (int) $swept : 0; + } + + /** + * Run sweep: open runs with no activity for 2× expected run wall-clock + * are marked `failed`; their pending/running stages are reaped. + * + * @return int Runs swept. + */ + private static function sweep_stuck_runs(): int { + if ( ! PRAutoBlogger_Run_State::is_available() ) { + return 0; + } + global $wpdb; + $runs = PRAutoBlogger_Run_State::table_name(); + + /** + * Filter the expected whole-run wall-clock seconds used by the + * stuck-run sweep (threshold is 2× this value). + * + * @param int $seconds Expected seconds (default 1800). + */ + $expected = (int) apply_filters( 'prautoblogger_filter_expected_run_seconds', self::EXPECTED_RUN_SECONDS ); + $cutoff = gmdate( 'Y-m-d H:i:s', time() - ( 2 * $expected ) ); + + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching + $stuck = $wpdb->get_col( + $wpdb->prepare( + "SELECT run_id FROM {$runs} WHERE status IN ('pending','running') AND updated_at < %s", + $cutoff + ) + ); + if ( ! is_array( $stuck ) || empty( $stuck ) ) { + return 0; + } + + $swept = 0; + foreach ( $stuck as $run_id ) { + $run_id = (string) $run_id; + if ( ! PRAutoBlogger_Run_State::mark_status( $run_id, 'failed' ) ) { + continue; + } + self::reap_open_stages( $run_id ); + PRAutoBlogger_Logger::instance()->info( + sprintf( 'Run %s stuck past 2x expected wall-clock — marked failed (renders as incomplete).', $run_id ), + 'run-reaper' + ); + ++$swept; + } + return $swept; + } + + /** + * Mark a failed run's pending/running stage rows as failed. + * + * @param string $run_id Run UUID. + * @return void + */ + private static function reap_open_stages( string $run_id ): void { + if ( ! PRAutoBlogger_Run_Stage_State::is_available() ) { + return; + } + global $wpdb; + $table = PRAutoBlogger_Run_Stage_State::table_name(); + $now = current_time( 'mysql' ); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $wpdb->query( + $wpdb->prepare( + "UPDATE {$table} SET status = 'failed', updated_at = %s, finished_at = %s + WHERE run_id = %s AND status IN ('pending','running')", + $now, + $now, + $run_id + ) + ); + } + + /** + * Retention: NULL heavy payloads older than R days. R is read from the + * `prautoblogger_request_json_retention_days` setting at every call + * (default = PRAUTOBLOGGER_DEFAULT_REQUEST_JSON_RETENTION_DAYS; + * 0 or negative = keep forever). + * + * @return int Rows pruned (both tables combined). + */ + private static function prune_payloads(): int { + $days = (int) get_option( + 'prautoblogger_request_json_retention_days', + PRAUTOBLOGGER_DEFAULT_REQUEST_JSON_RETENTION_DAYS + ); + if ( $days <= 0 ) { + return 0; + } + + global $wpdb; + $cutoff = gmdate( 'Y-m-d H:i:s', time() - ( $days * DAY_IN_SECONDS ) ); + $pruned = 0; + + $gen_log = $wpdb->prefix . 'prautoblogger_generation_log'; + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $result = $wpdb->query( + $wpdb->prepare( + "UPDATE {$gen_log} SET request_json = NULL WHERE request_json IS NOT NULL AND created_at < %s", + $cutoff + ) + ); + $pruned += is_numeric( $result ) ? (int) $result : 0; + + if ( PRAutoBlogger_Run_Stage_State::is_available() ) { + $stages = PRAutoBlogger_Run_Stage_State::table_name(); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $result = $wpdb->query( + $wpdb->prepare( + "UPDATE {$stages} SET meta_json = NULL WHERE meta_json IS NOT NULL AND updated_at < %s", + $cutoff + ) + ); + $pruned += is_numeric( $result ) ? (int) $result : 0; + } + + return $pruned; + } +} diff --git a/includes/core/class-run-stage-state.php b/includes/core/class-run-stage-state.php new file mode 100644 index 0000000..67eca31 --- /dev/null +++ b/includes/core/class-run-stage-state.php @@ -0,0 +1,293 @@ +prefix . 'prautoblogger_run_stages'; + } + + /** + * Whether the run_stages table exists and is queryable. Cached per + * request; never throws. + * + * @return bool + */ + public static function is_available(): bool { + if ( null !== self::$table_ok ) { + return self::$table_ok; + } + global $wpdb; + if ( null === $wpdb ) { + return false; // Not cached: $wpdb may appear later in boot. + } + $table = self::table_name(); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching + $found = $wpdb->get_var( $wpdb->prepare( 'SHOW TABLES LIKE %s', $table ) ); + self::$table_ok = ( $found === $table ); + return self::$table_ok; + } + + /** + * Stable hash identifying one article idea within a run (post-creation + * idempotency + stage item scoping both key off this). + * + * @param PRAutoBlogger_Article_Idea $idea The idea. + * @return string 16-hex-char hash. + */ + public static function idea_hash( PRAutoBlogger_Article_Idea $idea ): string { + return substr( md5( $idea->get_suggested_title() . '|' . $idea->get_topic() ), 0, 16 ); + } + + /** + * Stage item_key for an article idea ('' is reserved for run-level stages). + * + * @param PRAutoBlogger_Article_Idea $idea The idea. + * @return string e.g. 'idea:1a2b3c4d5e6f7a8b'. + */ + public static function item_key_for_idea( PRAutoBlogger_Article_Idea $idea ): string { + return 'idea:' . self::idea_hash( $idea ); + } + + /** + * Mark a stage as entered (running). Upsert: first entry inserts the + * row; re-entry bumps `attempt` — but a `done` stage is sticky and is + * neither demoted nor re-counted. + * + * Side effects: one INSERT … ON DUPLICATE KEY UPDATE. + * + * @param string $run_id Run UUID. + * @param string $stage Stage name (see Stage_Display_Map). + * @param string $agent_role Fan-out dimension ('' default in Phase 1). + * @param string $item_key Article scope ('' for run-level stages). + * @return void + */ + public static function start( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): void { + if ( '' === $run_id || ! self::is_available() ) { + return; + } + global $wpdb; + $table = self::table_name(); + $now = current_time( 'mysql' ); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $wpdb->query( + $wpdb->prepare( + "INSERT INTO {$table} + (run_id, stage, agent_role, item_key, status, attempt, started_at, updated_at) + VALUES (%s, %s, %s, %s, 'running', 1, %s, %s) + ON DUPLICATE KEY UPDATE + attempt = IF(status = 'done', attempt, attempt + 1), + status = IF(status = 'done', 'done', 'running'), + updated_at = VALUES(updated_at)", + $run_id, + $stage, + $agent_role, + $item_key, + $now, + $now + ) + ); + } + + /** + * Mark a stage done, persisting its output for resume-without-recharge. + * + * Side effects: one INSERT … ON DUPLICATE KEY UPDATE. + * + * @param string $run_id Run UUID. + * @param string $stage Stage name. + * @param string $agent_role Fan-out dimension. + * @param string $item_key Article scope. + * @param string|null $output Stage output to snapshot (pruned after R days). + * @param float $cost_usd Cost attributed to the stage. + * @return void + */ + public static function done( string $run_id, string $stage, string $agent_role = '', string $item_key = '', ?string $output = null, float $cost_usd = 0.0 ): void { + if ( '' === $run_id || ! self::is_available() ) { + return; + } + global $wpdb; + $table = self::table_name(); + $now = current_time( 'mysql' ); + $meta = null !== $output ? wp_json_encode( array( 'output' => $output ) ) : null; + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $wpdb->query( + $wpdb->prepare( + "INSERT INTO {$table} + (run_id, stage, agent_role, item_key, status, attempt, cost_usd, meta_json, started_at, updated_at, finished_at) + VALUES (%s, %s, %s, %s, 'done', 1, %f, %s, %s, %s, %s) + ON DUPLICATE KEY UPDATE + status = 'done', + cost_usd = VALUES(cost_usd), + meta_json = VALUES(meta_json), + updated_at = VALUES(updated_at), + finished_at = VALUES(finished_at)", + $run_id, + $stage, + $agent_role, + $item_key, + $cost_usd, + $meta, + $now, + $now, + $now + ) + ); + } + + /** + * Mark a stage failed (does not demote a `done` stage). + * + * @param string $run_id Run UUID. + * @param string $stage Stage name. + * @param string $agent_role Fan-out dimension. + * @param string $item_key Article scope. + * @return void + */ + public static function fail( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): void { + if ( '' === $run_id || ! self::is_available() ) { + return; + } + global $wpdb; + $table = self::table_name(); + $now = current_time( 'mysql' ); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $wpdb->query( + $wpdb->prepare( + "UPDATE {$table} SET status = 'failed', updated_at = %s, finished_at = %s + WHERE run_id = %s AND stage = %s AND agent_role = %s AND item_key = %s AND status != 'done'", + $now, + $now, + $run_id, + $stage, + $agent_role, + $item_key + ) + ); + } + + /** + * Fetch one stage row. + * + * @param string $run_id Run UUID. + * @param string $stage Stage name. + * @param string $agent_role Fan-out dimension. + * @param string $item_key Article scope. + * @return array|null Row or null. + */ + public static function get( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): ?array { + if ( '' === $run_id || ! self::is_available() ) { + return null; + } + global $wpdb; + $table = self::table_name(); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching + $row = $wpdb->get_row( + $wpdb->prepare( + "SELECT * FROM {$table} WHERE run_id = %s AND stage = %s AND agent_role = %s AND item_key = %s", + $run_id, + $stage, + $agent_role, + $item_key + ), + ARRAY_A + ); + return is_array( $row ) ? $row : null; + } + + /** + * Whether a stage already completed (resume entry-point check). + * + * @param string $run_id Run UUID. + * @param string $stage Stage name. + * @param string $agent_role Fan-out dimension. + * @param string $item_key Article scope. + * @return bool + */ + public static function is_done( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): bool { + $row = self::get( $run_id, $stage, $agent_role, $item_key ); + return null !== $row && 'done' === $row['status']; + } + + /** + * A done stage's stored output (null when absent, not done, or pruned). + * + * @param string $run_id Run UUID. + * @param string $stage Stage name. + * @param string $agent_role Fan-out dimension. + * @param string $item_key Article scope. + * @return string|null + */ + public static function get_output( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): ?string { + $row = self::get( $run_id, $stage, $agent_role, $item_key ); + if ( null === $row || 'done' !== $row['status'] || empty( $row['meta_json'] ) ) { + return null; + } + $meta = json_decode( (string) $row['meta_json'], true ); + return is_array( $meta ) && isset( $meta['output'] ) ? (string) $meta['output'] : null; + } + + /** + * Fail every still-open (pending/running) stage of one item — used by + * the worker's catch-all so a crashed article leaves no stage stuck + * in 'running' until the reaper. Done stages are untouched. + * + * @param string $run_id Run UUID. + * @param string $item_key Stage item key ('' = run-level stages). + * @return void + */ + public static function fail_open_for_item( string $run_id, string $item_key ): void { + if ( '' === $run_id || ! self::is_available() ) { + return; + } + global $wpdb; + $table = self::table_name(); + $now = current_time( 'mysql' ); + // phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery + $wpdb->query( + $wpdb->prepare( + "UPDATE {$table} SET status = 'failed', updated_at = %s, finished_at = %s + WHERE run_id = %s AND item_key = %s AND status IN ('pending','running')", + $now, + $now, + $run_id, + $item_key + ) + ); + } + + /** Reset per-request caches (tests). */ + public static function flush_cache(): void { + self::$table_ok = null; + } +} diff --git a/tests/unit/Core/RunReaperTest.php b/tests/unit/Core/RunReaperTest.php new file mode 100644 index 0000000..23877f2 --- /dev/null +++ b/tests/unit/Core/RunReaperTest.php @@ -0,0 +1,115 @@ +query(). + * + * @var string[] + */ + private array $queries = array(); + + protected function setUp(): void { + parent::setUp(); + $this->queries = array(); + // Local mock: the reaper also uses get_col(), which the shared + // BaseTestCase mock does not declare. + $this->wpdb = $this->getMockBuilder( \stdClass::class ) + ->addMethods( array( 'prepare', 'get_var', 'get_results', 'get_col', 'insert', 'query', 'get_row', 'update' ) ) + ->getMock(); + $this->wpdb->prefix = 'wp_'; + $this->wpdb->insert_id = 0; + $this->wpdb->last_error = ''; + $GLOBALS['wpdb'] = $this->wpdb; + $this->wpdb->method( 'prepare' )->willReturnCallback( + static function ( $sql, ...$args ) { + return $sql . ' /* ' . implode( ',', array_map( 'strval', $args ) ) . ' */'; + } + ); + Functions\when( 'apply_filters' )->returnArg( 2 ); + \PRAutoBlogger_Run_State::flush_cache(); + \PRAutoBlogger_Run_Stage_State::flush_cache(); + } + + protected function tearDown(): void { + \PRAutoBlogger_Run_State::flush_cache(); + \PRAutoBlogger_Run_Stage_State::flush_cache(); + unset( $GLOBALS['wpdb'] ); + parent::tearDown(); + } + + /** + * Retention 0 (or negative) means "keep forever": the pruner must not + * issue any UPDATE. + */ + public function test_zero_retention_keeps_payloads_forever(): void { + $this->stub_get_option( + array( + 'prautoblogger_request_json_retention_days' => 0, + ) + ); + $this->wpdb->method( 'get_var' )->willReturn( null ); // No state tables. + $this->wpdb->method( 'get_col' )->willReturn( array() ); + $this->wpdb->expects( $this->never() )->method( 'query' ); + + $stats = \PRAutoBlogger_Run_Reaper::reap(); + $this->assertSame( 0, $stats['payloads_pruned'] ); + } + + /** + * With the default setting the pruner NULLs request_json older than R + * days on the generation log (and stage payloads when that table + * exists). The sweeps only touch 'running'/open rows. + */ + public function test_reap_prunes_and_sweeps_with_setting_backed_window(): void { + $this->stub_get_option( + array( + 'prautoblogger_request_json_retention_days' => PRAUTOBLOGGER_DEFAULT_REQUEST_JSON_RETENTION_DAYS, + ) + ); + // Both state tables exist. + $this->wpdb->method( 'get_var' )->willReturnCallback( + static function ( $sql ) { + if ( false !== strpos( (string) $sql, 'run_stages' ) ) { + return 'wp_prautoblogger_run_stages'; + } + return 'wp_prautoblogger_runs'; + } + ); + $this->wpdb->method( 'get_col' )->willReturn( array() ); // No stuck runs. + $this->wpdb->method( 'query' )->willReturnCallback( + function ( $sql ) { + $this->queries[] = (string) $sql; + return 2; + } + ); + + $stats = \PRAutoBlogger_Run_Reaper::reap(); + + $all_sql = implode( "\n", $this->queries ); + $this->assertStringContainsString( 'request_json = NULL', $all_sql ); + $this->assertStringContainsString( 'meta_json = NULL', $all_sql ); + $this->assertStringContainsString( "status = 'failed'", $all_sql ); + $this->assertStringContainsString( "WHERE status = 'running' AND updated_at <", $all_sql ); + $this->assertSame( 4, $stats['payloads_pruned'] ); // 2 rows x 2 tables. + } +} diff --git a/tests/unit/Core/RunStageStateTest.php b/tests/unit/Core/RunStageStateTest.php new file mode 100644 index 0000000..254ef2b --- /dev/null +++ b/tests/unit/Core/RunStageStateTest.php @@ -0,0 +1,140 @@ +query(). + * + * @var string[] + */ + private array $queries = array(); + + protected function setUp(): void { + parent::setUp(); + $this->queries = array(); + $this->wpdb = $this->create_mock_wpdb(); + $GLOBALS['wpdb'] = $this->wpdb; + $this->wpdb->method( 'prepare' )->willReturnCallback( + static function ( $sql, ...$args ) { + return $sql . ' /* ' . implode( ',', array_map( 'strval', $args ) ) . ' */'; + } + ); + \PRAutoBlogger_Run_Stage_State::flush_cache(); + } + + protected function tearDown(): void { + \PRAutoBlogger_Run_Stage_State::flush_cache(); + unset( $GLOBALS['wpdb'] ); + parent::tearDown(); + } + + /** + * Two different ideas in the same run must get different item keys + * (one run_id spans all N articles of a batch run); the same idea must + * hash stably across processes (queue serialization round-trip). + */ + public function test_item_keys_scope_articles_within_a_run(): void { + $idea_a = new \PRAutoBlogger_Article_Idea( + array( + 'topic' => 'BPC-157 dosing', + 'article_type' => 'guide', + 'suggested_title' => 'BPC-157 Dosing Guide', + 'summary' => 's', + ) + ); + $idea_b = new \PRAutoBlogger_Article_Idea( + array( + 'topic' => 'TB-500 storage', + 'article_type' => 'question', + 'suggested_title' => 'How To Store TB-500', + 'summary' => 's', + ) + ); + $idea_a_again = new \PRAutoBlogger_Article_Idea( $idea_a->to_array() ); + + $key_a = \PRAutoBlogger_Run_Stage_State::item_key_for_idea( $idea_a ); + $key_b = \PRAutoBlogger_Run_Stage_State::item_key_for_idea( $idea_b ); + + $this->assertNotSame( $key_a, $key_b ); + $this->assertSame( $key_a, \PRAutoBlogger_Run_Stage_State::item_key_for_idea( $idea_a_again ) ); + $this->assertStringStartsWith( 'idea:', $key_a ); + $this->assertLessThanOrEqual( 64, strlen( $key_a ) ); + $this->assertSame( + \PRAutoBlogger_Run_Stage_State::idea_hash( $idea_a ), + substr( $key_a, 5 ) + ); + } + + /** + * start() upserts must never demote a done stage and must bump attempt + * on re-entry; done() must be sticky. + */ + public function test_start_upsert_protects_done_status(): void { + $this->wpdb->method( 'get_var' )->willReturn( 'wp_prautoblogger_run_stages' ); + $this->wpdb->method( 'query' )->willReturnCallback( + function ( $sql ) { + $this->queries[] = (string) $sql; + return 1; + } + ); + + \PRAutoBlogger_Run_Stage_State::start( 'run-1', 'draft', '', 'idea:abc' ); + + $sql = implode( "\n", $this->queries ); + $this->assertStringContainsString( 'ON DUPLICATE KEY UPDATE', $sql ); + $this->assertStringContainsString( "IF(status = 'done', attempt, attempt + 1)", $sql ); + $this->assertStringContainsString( "IF(status = 'done', 'done', 'running')", $sql ); + } + + /** + * fail() and fail_open_for_item() must never touch done rows. + */ + public function test_fail_paths_exclude_done_rows(): void { + $this->wpdb->method( 'get_var' )->willReturn( 'wp_prautoblogger_run_stages' ); + $this->wpdb->method( 'query' )->willReturnCallback( + function ( $sql ) { + $this->queries[] = (string) $sql; + return 1; + } + ); + + \PRAutoBlogger_Run_Stage_State::fail( 'run-1', 'draft', '', 'idea:abc' ); + \PRAutoBlogger_Run_Stage_State::fail_open_for_item( 'run-1', 'idea:abc' ); + + $this->assertStringContainsString( "status != 'done'", $this->queries[0] ); + $this->assertStringContainsString( "status IN ('pending','running')", $this->queries[1] ); + } + + /** + * Without the table (half-migrated site) every method degrades to a + * harmless no-op/false/null — never a fatal. + */ + public function test_no_table_degrades_to_noops(): void { + $this->wpdb->method( 'get_var' )->willReturn( null ); // SHOW TABLES miss. + $this->wpdb->expects( $this->never() )->method( 'query' ); + + \PRAutoBlogger_Run_Stage_State::start( 'run-1', 'draft' ); + \PRAutoBlogger_Run_Stage_State::done( 'run-1', 'draft', '', '', 'output' ); + \PRAutoBlogger_Run_Stage_State::fail( 'run-1', 'draft' ); + $this->assertFalse( \PRAutoBlogger_Run_Stage_State::is_done( 'run-1', 'draft' ) ); + $this->assertNull( \PRAutoBlogger_Run_Stage_State::get_output( 'run-1', 'draft' ) ); + } +}