peptide-repo/prautoblogger
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(M3): edit+rerun schema substrate — stage_inputs store, stale/human_modified columns, db 1.3.0

Browse Source 
- Pipeline_Schema_Installer: run_stages gains human_modified + stale
  (stale is a COLUMN not a status: rows stay 'done' so resume code can
  never silently re-run them — CPO guardrail 3 by construction);
  run_decisions gains human_modified; new INSERT-only stage_inputs table
  (edit forks as immutable versions + idea seeds for re-run-from-here).
  All dbDelta-idempotent; PRAUTOBLOGGER_DB_VERSION 1.1.0 -> 1.3.0 (the
  constant never carried the documented 1.2.0) so the existing
  admin_init/cron self-healing migration path fires everywhere.
- Run_Stage_State split (M2 pattern, was 313 lines): pipeline write
  bodies moved verbatim to Run_Stage_Writes (public API unchanged via
  thin proxies); done() now clears stale on fresh output with a
  half-migrated-schema retry (Cost_Tracker v0.18.0 pattern); new
  item-scoped read stages_for_item(). Operator-action mutations
  (restart/mark_stale/demote_to_pending) isolated in
  Run_Stage_Rerun_State — the only code allowed to demote a done row.
- Audit_Writer::flag_decisions_human_modified() (guardrail 2 on
  run_decisions). Run_Reaper retention prunes human fork bodies (seeds
  kept). Uninstall covers the new table + rerun cron hooks.

Agent-Session: engineer-prab-m3-20260612
This commit is contained in:
peptiderepo
2026-06-12 13:05:55 +00:00
parent 1b31ef6396
commit aaf1b4274e
11 changed files with 1137 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
includes/core/class-audit-writer.php
View File

@@ -125,6 +125,37 @@ class PRAutoBlogger_Audit_Writer {
}
}
/**
* Flag the run_decisions rows of given stages as human-modified
* (v0.20.0 / CPO guardrail 2). Used in two cases: the decision rows of
* a stage whose edited input enters execution, and decisions recorded
* during a re-run-from-here window on an item that carries a human
* edit (derived-from-edited-content). Self-healing no-op when the
* column/table is missing.
*
* @param string $run_id Run UUID.
* @param array<string> $stages Stage names to flag.
* @param string|null $since Optional MySQL datetime — only flag rows created at/after it.
* @return int Rows flagged.
*/
public static function flag_decisions_human_modified( string $run_id, array $stages, ?string $since = null ): int {
if ( '' === $run_id || empty( $stages ) || ! self::is_available() ) {
return 0;
}
global $wpdb;
$table = $wpdb->prefix . 'prautoblogger_run_decisions';
$placeholders = implode( ',', array_fill( 0, count( $stages ), '%s' ) );
$sql = "UPDATE {$table} SET human_modified = 1 WHERE run_id = %s AND stage IN ({$placeholders})";
$params = array_merge( array( $run_id ), array_values( $stages ) );
if ( null !== $since ) {
$sql .= ' AND created_at >= %s';
$params[] = $since;
}
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.PreparedSQL.NotPrepared -- $placeholders is a fixed %s list.
$updated = $wpdb->query( $wpdb->prepare( $sql, $params ) );
return is_numeric( $updated ) ? (int) $updated : 0;
}
/** Reset per-request caches (tests). */
public static function flush_cache(): void {
self::$tables_ok = null;

8
includes/core/class-run-reaper.php
View File

@@ -16,8 +16,9 @@ declare(strict_types=1);
* ledger/state activity for 2× the expected run wall-clock are
* marked `failed` and their non-done stages reaped to `failed`;
* such runs render as "incomplete" via incomplete_runs().
* 3. Retention — `request_json` on generation_log and stage output
* 3. Retention — `request_json` on generation_log, stage output
* payloads (`meta_json` on run_stages) are NULLed after R days;
* payloads, and stage_inputs human-fork bodies (v0.20.0);
* R comes from the `prautoblogger_request_json_retention_days`
* SETTING (default constant 14; 0 = keep forever) — never a
* literal at the point of use.
@@ -265,6 +266,11 @@ class PRAutoBlogger_Run_Reaper {
$pruned += is_numeric( $result ) ? (int) $result : 0;
}
// v0.20.0: human edit-fork bodies ride the same retention (seed
// rows are kept — structural, ~1KB). A pruned fork can no longer
// be replayed; the dossier explains that state honestly.
$pruned += PRAutoBlogger_Stage_Input_Store::prune_human_bodies( $cutoff );
return $pruned;
}
}

148
includes/core/class-run-stage-rerun-state.php Normal file
View File

@@ -0,0 +1,148 @@
<?php
declare(strict_types=1);
/**
* phpcs:ignore WordPress.Files.FileName.InvalidClassFileName -- class naming convention differs from WordPress standard
*
* Operator-action state mutations for edit + re-run (v0.20.0, M3).
*
* What: The ONLY paths that may demote a 'done' run_stages row — and all
* of them are deliberate operator actions dispatched from the
* dossier (never reachable from pipeline resume code):
* - restart(): done/failed/halted/pending → running for a single-
* stage replay; bumps attempt, clears stale, sets human_modified
* when the replay uses an edited input fork (CPO guardrail 2).
* - mark_stale(): flags downstream 'done' stages as stale after an
* upstream edit enters execution. Stale is a COLUMN, not a
* status: the stage stays 'done', so resume-without-recharge
* logic can never silently re-run it (CPO guardrail 3 holds by
* construction). Only Run_Stage_Writes::done() clears it.
* - demote_to_pending(): re-run-from-here demotion; the worker
* re-entry then rebuilds those stages from current upstream
* snapshots. Audit columns (attempt, human_modified) survive;
* old meta_json survives until overwritten by the fresh done().
* Every method no-ops when the run_stages table is missing.
* Who triggers it: PRAutoBlogger_Rerun_Executor (cron handlers, after
* re-validating eligibility under the generation lock).
* Dependencies: Run_Stage_State (probe/table), $wpdb.
*
* @see core/class-rerun-executor.php — Sole caller.
* @see core/class-run-stage-writes.php — Pipeline-path writes (done clears stale).
* @see core/class-rerun-eligibility.php — Policy gates before any mutation.
* @see ARCHITECTURE.md #24 — Edit + re-run design.
*/
class PRAutoBlogger_Run_Stage_Rerun_State {
/**
* Demote one stage row to 'running' for an operator replay.
*
* Allowed from done/failed/halted/pending — never from 'running'
* (a concurrently executing stage cannot be restarted). Bumps the
* attempt counter, clears `stale` (the row is being refreshed) and
* sets `human_modified` when the replay executes an edited fork.
* `human_modified` is never cleared by any code path — it is sticky
* audit state for the life of the run.
*
* Side effects: one UPDATE on run_stages.
*
* @param string $run_id Run UUID.
* @param string $stage Stage name.
* @param string $agent_role Agent role (resolved by the caller from the actual row).
* @param string $item_key Item key.
* @param bool $human_modified Whether an edited input fork drives this replay.
* @return bool Whether a row was demoted.
*/
public static function restart( string $run_id, string $stage, string $agent_role, string $item_key, bool $human_modified ): bool {
if ( '' === $run_id || ! PRAutoBlogger_Run_Stage_State::is_available() ) {
return false;
}
global $wpdb;
$table = PRAutoBlogger_Run_Stage_State::table_name();
$now = current_time( 'mysql' );
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery
$updated = $wpdb->query(
$wpdb->prepare(
"UPDATE {$table}
SET status = 'running', attempt = attempt + 1, stale = 0,
human_modified = IF(%d = 1, 1, human_modified),
updated_at = %s, finished_at = NULL
WHERE run_id = %s AND stage = %s AND agent_role = %s AND item_key = %s
AND status IN ('done','failed','halted','pending')",
$human_modified ? 1 : 0,
$now,
$run_id,
$stage,
$agent_role,
$item_key
)
);
return 1 === (int) $updated;
}
/**
* Flag a set of 'done' stages as stale (explicit downstream
* invalidation after an upstream edit/re-run — CPO guardrail 3).
*
* Nothing is demoted and nothing auto-re-runs: the rows stay 'done'
* and visible; the operator triggers each subsequent stage (or
* re-run-from-here) deliberately.
*
* @param string $run_id Run UUID.
* @param string $item_key Item key.
* @param array<string> $stages Stage names to flag.
* @return int Rows flagged.
*/
public static function mark_stale( string $run_id, string $item_key, array $stages ): int {
if ( '' === $run_id || empty( $stages ) || ! PRAutoBlogger_Run_Stage_State::is_available() ) {
return 0;
}
global $wpdb;
$table = PRAutoBlogger_Run_Stage_State::table_name();
$placeholders = implode( ',', array_fill( 0, count( $stages ), '%s' ) );
$params = array_merge( array( current_time( 'mysql' ), $run_id, $item_key ), array_values( $stages ) );
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.PreparedSQL.NotPrepared -- $placeholders is a fixed %s list.
$updated = $wpdb->query(
$wpdb->prepare(
"UPDATE {$table} SET stale = 1, updated_at = %s
WHERE run_id = %s AND item_key = %s AND status = 'done' AND stage IN ({$placeholders})",
$params
)
);
return is_numeric( $updated ) ? (int) $updated : 0;
}
/**
* Demote a set of stages to 'pending' for re-run-from-here.
*
* The pipeline's resume machinery treats pending stages as not-done
* and re-executes them with prompts REBUILT from current upstream
* snapshots — that is how an upstream edit propagates downstream.
* Rows keep attempt/human_modified (audit) and their old meta_json
* until the fresh completion overwrites it; `stale` stays 1 until
* Run_Stage_Writes::done() clears it on fresh output. Never touches
* a 'running' row.
*
* @param string $run_id Run UUID.
* @param string $item_key Item key.
* @param array<string> $stages Stage names to demote.
* @return int Rows demoted.
*/
public static function demote_to_pending( string $run_id, string $item_key, array $stages ): int {
if ( '' === $run_id || empty( $stages ) || ! PRAutoBlogger_Run_Stage_State::is_available() ) {
return 0;
}
global $wpdb;
$table = PRAutoBlogger_Run_Stage_State::table_name();
$placeholders = implode( ',', array_fill( 0, count( $stages ), '%s' ) );
$params = array_merge( array( current_time( 'mysql' ), $run_id, $item_key ), array_values( $stages ) );
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.PreparedSQL.NotPrepared -- $placeholders is a fixed %s list.
$updated = $wpdb->query(
$wpdb->prepare(
"UPDATE {$table} SET status = 'pending', stale = 1, updated_at = %s, finished_at = NULL
WHERE run_id = %s AND item_key = %s AND status != 'running' AND stage IN ({$placeholders})",
$params
)
);
return is_numeric( $updated ) ? (int) $updated : 0;
}
}

150
includes/core/class-run-stage-state.php
View File

@@ -16,9 +16,17 @@ declare(strict_types=1);
* roles (Phase-2 fan-out) are honoured as-is. get() falls back to role=''
* on miss for mid-run-upgrade compat with v0.18.1 rows.
*
* v0.20.0: write bodies extracted to PRAutoBlogger_Run_Stage_Writes (the
* M2 split pattern — this class was at the 300-line cap); the public API
* here is unchanged and delegates. Operator-action mutations (replay
* restart, stale marking, re-run demotion) live in
* PRAutoBlogger_Run_Stage_Rerun_State only.
*
* @see core/class-run-stage-writes.php — Extracted write bodies.
* @see core/class-run-stage-rerun-state.php — M3 operator-action writes.
* @see core/class-run-state.php — Run-level lifecycle + ledger row.
* @see core/class-run-reaper.php — Sweeps stages stuck in 'running'.
* @see ARCHITECTURE.md #22 — Pipeline v2 Phase 1 substrate.
* @see ARCHITECTURE.md #22 / #24 — Substrate + edit/re-run design.
*/
class PRAutoBlogger_Run_Stage_State {
@@ -73,6 +81,7 @@ class PRAutoBlogger_Run_Stage_State {
/**
* Mark a stage as running. Sticky: done stages are never demoted.
* Delegates to Run_Stage_Writes (v0.20.0 split — behavior unchanged).
*
* @param string $run_id Run UUID.
* @param string $stage Stage name (see Stage_Display_Map).
@@ -80,35 +89,13 @@ class PRAutoBlogger_Run_Stage_State {
* @param string $item_key Article scope ('' for run-level stages).
*/
public static function start( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): void {
if ( '' === $run_id || ! self::is_available() ) {
return;
}
$agent_role = self::resolve_role( $stage, $agent_role );
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
)
);
PRAutoBlogger_Run_Stage_Writes::start( $run_id, $stage, $agent_role, $item_key );
}
/**
* Mark a stage done and persist its output for resume-without-recharge.
* Delegates to Run_Stage_Writes (v0.20.0 split; fresh completion also
* clears the `stale` flag there).
*
* @param string $run_id Run UUID.
* @param string $stage Stage name.
@@ -118,41 +105,12 @@ class PRAutoBlogger_Run_Stage_State {
* @param float $cost_usd Cost attributed to this stage.
*/
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;
}
$agent_role = self::resolve_role( $stage, $agent_role );
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
)
);
PRAutoBlogger_Run_Stage_Writes::done( $run_id, $stage, $agent_role, $item_key, $output, $cost_usd );
}
/**
* Mark a stage failed (does not demote a done stage).
* Mark a stage failed (does not demote a done stage). Delegates to
* Run_Stage_Writes (v0.20.0 split — behavior unchanged).
*
* @param string $run_id Run UUID.
* @param string $stage Stage name.
@@ -160,26 +118,7 @@ class PRAutoBlogger_Run_Stage_State {
* @param string $item_key Article scope.
*/
public static function fail( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): void {
if ( '' === $run_id || ! self::is_available() ) {
return;
}
$agent_role = self::resolve_role( $stage, $agent_role );
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
)
);
PRAutoBlogger_Run_Stage_Writes::fail( $run_id, $stage, $agent_role, $item_key );
}
/**
@@ -196,7 +135,7 @@ class PRAutoBlogger_Run_Stage_State {
if ( '' === $run_id || ! self::is_available() ) {
return null;
}
$agent_role = self::resolve_role( $stage, $agent_role );
$agent_role = PRAutoBlogger_Run_Stage_Writes::resolve_role( $stage, $agent_role );
global $wpdb;
$table = self::table_name();
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
@@ -265,49 +204,42 @@ class PRAutoBlogger_Run_Stage_State {
/**
* Fail every still-open stage of one item. Done stages are untouched.
* Delegates to Run_Stage_Writes (v0.20.0 split — behavior unchanged).
*
* @param string $run_id Run UUID.
* @param string $item_key Stage item key ('' = run-level stages).
*/
public static function fail_open_for_item( string $run_id, string $item_key ): void {
PRAutoBlogger_Run_Stage_Writes::fail_open_for_item( $run_id, $item_key );
}
/**
* All stage rows for one item of a run (plus run-level rows when
* $include_run_level), ordered by id. Read API for the rerun engine
* and the dossier's item-scoped assembly (v0.20.0).
*
* @param string $run_id Run UUID.
* @param string $item_key Item key ('idea:<hash>').
* @param bool $include_run_level Include item_key='' rows too.
* @return array<int, array<string, mixed>> Stage rows (ARRAY_A).
*/
public static function stages_for_item( string $run_id, string $item_key, bool $include_run_level = true ): array {
if ( '' === $run_id || ! self::is_available() ) {
return;
return array();
}
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
)
);
$sql = $include_run_level
? "SELECT * FROM {$table} WHERE run_id = %s AND item_key IN (%s, '') ORDER BY id ASC"
: "SELECT * FROM {$table} WHERE run_id = %s AND item_key = %s ORDER BY id ASC";
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching,WordPress.DB.PreparedSQL.NotPrepared -- assembled from two fixed literals above.
$rows = $wpdb->get_results( $wpdb->prepare( $sql, $run_id, $item_key ), ARRAY_A );
return is_array( $rows ) ? $rows : array();
}
/** Reset per-request caches (tests). */
/** Reset per-request caches (tests). */ /** Reset per-request caches (tests). */
public static function flush_cache(): void {
self::$table_ok = null;
}
/**
* Resolve the effective agent role for a stage.
*
* '' → derive from Stage_Display_Map (unknown stages map to '').
* Non-empty → returned as-is (Phase-2 explicit fan-out role).
*
* @param string $stage Stage name.
* @param string $agent_role Caller-supplied role ('' = auto-resolve).
* @return string Effective agent role.
*/
private static function resolve_role( string $stage, string $agent_role ): string {
if ( '' !== $agent_role ) {
return $agent_role;
}
return (string) PRAutoBlogger_Stage_Display_Map::default_agent_role( $stage );
}
}

222
includes/core/class-run-stage-writes.php Normal file
View File

@@ -0,0 +1,222 @@
<?php
declare(strict_types=1);
/**
* phpcs:ignore WordPress.Files.FileName.InvalidClassFileName -- class naming convention differs from WordPress standard
*
* Pipeline-path write operations over `wp_prautoblogger_run_stages`.
*
* Extracted from PRAutoBlogger_Run_Stage_State in v0.20.0 so that class
* stays under the 300-line cap (the M2 class-split pattern): start/done/
* fail/fail_open_for_item bodies moved here verbatim; Run_Stage_State
* keeps thin delegating proxies, so every existing call site is
* unchanged. One M3 behavior delta, isolated in done(): a freshly
* completed stage clears its `stale` flag (fresh output is never stale),
* with a self-healing legacy retry for half-migrated schemas that lack
* the v0.20.0 columns (same retry pattern as Cost_Tracker v0.18.0).
*
* Operator-action mutations (restart/mark_stale/demote) live in
* PRAutoBlogger_Run_Stage_Rerun_State, not here.
*
* Triggered by: PRAutoBlogger_Run_Stage_State proxies (pipeline call sites).
* Dependencies: Run_Stage_State (table probe), Stage_Display_Map (role).
*
* @see core/class-run-stage-state.php — Public API + read methods.
* @see core/class-run-stage-rerun-state.php — M3 operator-action writes.
* @see ARCHITECTURE.md #22 / #24 — Substrate + edit/re-run design.
*/
class PRAutoBlogger_Run_Stage_Writes {
/**
* Mark a stage as running. Sticky: done stages are never demoted.
*
* @param string $run_id Run UUID.
* @param string $stage Stage name (see Stage_Display_Map).
* @param string $agent_role Fan-out dimension ('' = resolve from stage map).
* @param string $item_key Article scope ('' for run-level stages).
*/
public static function start( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): void {
if ( '' === $run_id || ! PRAutoBlogger_Run_Stage_State::is_available() ) {
return;
}
$agent_role = self::resolve_role( $stage, $agent_role );
global $wpdb;
$table = PRAutoBlogger_Run_Stage_State::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 and persist its output for resume-without-recharge.
*
* v0.20.0: a fresh completion clears the `stale` flag (and never
* touches `human_modified` — that flag is sticky run audit). When the
* site's schema predates the v0.20.0 columns (cron fired before the
* admin_init migration pass), the write self-heals by retrying the
* pre-v0.20.0 statement instead of losing the checkpoint.
*
* @param string $run_id Run UUID.
* @param string $stage Stage name.
* @param string $agent_role Fan-out dimension ('' = resolve from stage map).
* @param string $item_key Article scope.
* @param string|null $output Stage output snapshot (pruned after R days).
* @param float $cost_usd Cost attributed to this stage.
*/
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 || ! PRAutoBlogger_Run_Stage_State::is_available() ) {
return;
}
$agent_role = self::resolve_role( $stage, $agent_role );
global $wpdb;
$table = PRAutoBlogger_Run_Stage_State::table_name();
$now = current_time( 'mysql' );
$meta = null !== $output ? wp_json_encode( array( 'output' => $output ) ) : null;
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery
$result = $wpdb->query(
$wpdb->prepare(
"INSERT INTO {$table}
(run_id, stage, agent_role, item_key, status, attempt, cost_usd, meta_json, stale, started_at, updated_at, finished_at)
VALUES (%s, %s, %s, %s, 'done', 1, %f, %s, 0, %s, %s, %s)
ON DUPLICATE KEY UPDATE
status = 'done',
cost_usd = VALUES(cost_usd),
meta_json = VALUES(meta_json),
stale = 0,
updated_at = VALUES(updated_at),
finished_at = VALUES(finished_at)",
$run_id,
$stage,
$agent_role,
$item_key,
$cost_usd,
$meta,
$now,
$now,
$now
)
);
// Self-healing on a half-migrated schema (no `stale` column yet):
// retry with the pre-v0.20.0 statement rather than losing the
// resume checkpoint. Same pattern as Cost_Tracker (v0.18.0).
if ( false === $result ) {
// 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 ('' = resolve from stage map).
* @param string $item_key Article scope.
*/
public static function fail( string $run_id, string $stage, string $agent_role = '', string $item_key = '' ): void {
if ( '' === $run_id || ! PRAutoBlogger_Run_Stage_State::is_available() ) {
return;
}
$agent_role = self::resolve_role( $stage, $agent_role );
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 stage = %s AND agent_role = %s AND item_key = %s AND status != 'done'",
$now,
$now,
$run_id,
$stage,
$agent_role,
$item_key
)
);
}
/**
* Fail every still-open stage of one item. Done stages are untouched.
*
* @param string $run_id Run UUID.
* @param string $item_key Stage item key ('' = run-level stages).
*/
public static function fail_open_for_item( string $run_id, string $item_key ): void {
if ( '' === $run_id || ! 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 item_key = %s AND status IN ('pending','running')",
$now,
$now,
$run_id,
$item_key
)
);
}
/**
* Resolve the effective agent role for a stage.
*
* '' → derive from Stage_Display_Map (unknown stages map to '').
* Non-empty → returned as-is (Phase-2 explicit fan-out role).
*
* Public in v0.20.0 (was private on Run_Stage_State) so the read API
* and the rerun classes share one resolution source.
*
* @param string $stage Stage name.
* @param string $agent_role Caller-supplied role ('' = auto-resolve).
* @return string Effective agent role.
*/
public static function resolve_role( string $stage, string $agent_role ): string {
if ( '' !== $agent_role ) {
return $agent_role;
}
return (string) PRAutoBlogger_Stage_Display_Map::default_agent_role( $stage );
}
}

269
includes/core/class-stage-input-store.php Normal file
View File

@@ -0,0 +1,269 @@
<?php
declare(strict_types=1);
/**
* phpcs:ignore WordPress.Files.FileName.InvalidClassFileName -- class naming convention differs from WordPress standard
*
* INSERT-only store over `wp_prautoblogger_stage_inputs` (v0.20.0, M3).
*
* What: Immutable input-version rows for the edit + re-run feature.
* Two row kinds, discriminated by `source`:
* - 'human': an operator's edited request body for one stage. Each
* save creates the NEXT version for the (run, stage, role, item)
* scope — there is deliberately no UPDATE path in this class, so
* originals (and earlier forks) can never be overwritten (CPO
* guardrail 1; the executed original lives untouched in
* generation_log.request_json).
* - 'seed': the serialized Article_Idea persisted at worker start
* (stage='', version=1). Re-run-from-here reconstructs the EXACT
* idea from it — rebuilding the idea from post fields would risk
* an item_key hash mismatch (post_title is sanitized at insert)
* and a duplicate post.
* Self-healing: every method no-ops (null/false/empty) when the
* table is missing (half-migrated schema).
* Retention: human fork bodies are NULLed by the run-reaper after
* R days (same setting as all heavy payloads); seed rows are ~1KB
* structural state and are kept.
* Who triggers it: Dossier_Actions (save fork), Rerun_Executor (read fork
* / seed), Article_Worker (save seed), Run_Reaper (prune).
* Dependencies: WordPress $wpdb.
*
* @see admin/class-dossier-actions.php — Fork save endpoint.
* @see core/class-rerun-executor.php — Fork/seed consumption.
* @see core/class-run-reaper.php — Retention prune.
* @see ARCHITECTURE.md #24 — Edit + re-run design.
*/
class PRAutoBlogger_Stage_Input_Store {
/** Source discriminator: operator edit fork. */
public const SOURCE_HUMAN = 'human';
/** Source discriminator: idea seed persisted at worker start. */
public const SOURCE_SEED = 'seed';
/** @var bool|null Per-request "stage_inputs table exists" probe result. */
private static ?bool $table_ok = null;
/** @return string Fully-qualified stage_inputs table name. */
public static function table_name(): string {
global $wpdb;
return $wpdb->prefix . 'prautoblogger_stage_inputs';
}
/**
* Whether the stage_inputs table exists. Cached per request.
*
* @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;
}
/**
* Save a human edit fork as the next version for its scope.
*
* INSERT-only by design. The version number is MAX(version)+1 for the
* scope; a concurrent-save collision on the UNIQUE key is retried
* once with a refreshed version.
*
* Side effects: one (rarely two) INSERTs into stage_inputs.
*
* @param string $run_id Run UUID.
* @param string $stage Stage name.
* @param string $agent_role Stage row's agent role.
* @param string $item_key Stage row's item key.
* @param string $request_json Edited request body (full JSON, no headers).
* @param string $author WP user login of the editor.
* @return int|null New version number, or null on failure/missing table.
*/
public static function save_fork( string $run_id, string $stage, string $agent_role, string $item_key, string $request_json, string $author ): ?int {
if ( '' === $run_id || '' === $stage || ! self::is_available() ) {
return null;
}
for ( $attempt = 0; $attempt < 2; $attempt++ ) {
$version = self::next_version( $run_id, $stage, $agent_role, $item_key );
$inserted = self::insert_row( $run_id, $stage, $agent_role, $item_key, $version, self::SOURCE_HUMAN, $request_json, $author );
if ( $inserted ) {
return $version;
}
}
return null;
}
/**
* Latest human fork row for a scope, or null when none exist.
*
* @param string $run_id Run UUID.
* @param string $stage Stage name.
* @param string $agent_role Agent role.
* @param string $item_key Item key.
* @return array<string, mixed>|null Row incl. version/request_json/author/created_at.
*/
public static function latest_fork( 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 AND source = %s
ORDER BY version DESC LIMIT 1",
$run_id,
$stage,
$agent_role,
$item_key,
self::SOURCE_HUMAN
),
ARRAY_A
);
return is_array( $row ) ? $row : null;
}
/**
* Persist the idea seed for one run item (INSERT-once; replays no-op).
*
* @param string $run_id Run UUID.
* @param string $item_key Item key from Run_Stage_State::item_key_for_idea().
* @param string $idea_json Serialized Article_Idea::to_array() payload.
* @return void
*/
public static function save_seed( string $run_id, string $item_key, string $idea_json ): void {
if ( '' === $run_id || '' === $item_key || ! self::is_available() ) {
return;
}
if ( null !== self::get_seed( $run_id, $item_key ) ) {
return; // Seed already persisted (idempotent resume).
}
self::insert_row( $run_id, '', '', $item_key, 1, self::SOURCE_SEED, $idea_json, 'pipeline' );
}
/**
* The idea seed JSON for one run item, or null when absent.
*
* @param string $run_id Run UUID.
* @param string $item_key Item key.
* @return string|null
*/
public static function get_seed( string $run_id, string $item_key ): ?string {
if ( '' === $run_id || ! self::is_available() ) {
return null;
}
global $wpdb;
$table = self::table_name();
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
$json = $wpdb->get_var(
$wpdb->prepare(
"SELECT request_json FROM {$table}
WHERE run_id = %s AND stage = '' AND item_key = %s AND source = %s LIMIT 1",
$run_id,
$item_key,
self::SOURCE_SEED
)
);
return ( is_string( $json ) && '' !== $json ) ? $json : null;
}
/**
* Retention: NULL human fork bodies older than the cutoff. Seed rows
* are kept (structural, ~1KB). Version/author/flag metadata survives
* as the permanent audit trail.
*
* @param string $cutoff MySQL datetime; rows created before it are pruned.
* @return int Rows pruned.
*/
public static function prune_human_bodies( string $cutoff ): int {
if ( ! self::is_available() ) {
return 0;
}
global $wpdb;
$table = self::table_name();
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery
$result = $wpdb->query(
$wpdb->prepare(
"UPDATE {$table} SET request_json = NULL
WHERE source = %s AND request_json IS NOT NULL AND created_at < %s",
self::SOURCE_HUMAN,
$cutoff
)
);
return is_numeric( $result ) ? (int) $result : 0;
}
/** Reset per-request caches (tests). */
public static function flush_cache(): void {
self::$table_ok = null;
}
/**
* Next version number for a scope (1 when none exist).
*
* @param string $run_id Run UUID.
* @param string $stage Stage name.
* @param string $agent_role Agent role.
* @param string $item_key Item key.
* @return int
*/
private static function next_version( string $run_id, string $stage, string $agent_role, string $item_key ): int {
global $wpdb;
$table = self::table_name();
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery,WordPress.DB.DirectDatabaseQuery.NoCaching
$max = $wpdb->get_var(
$wpdb->prepare(
"SELECT MAX(version) FROM {$table}
WHERE run_id = %s AND stage = %s AND agent_role = %s AND item_key = %s",
$run_id,
$stage,
$agent_role,
$item_key
)
);
return ( null !== $max ? (int) $max : 0 ) + 1;
}
/**
* Raw row INSERT (the only write path — INSERT-only by construction).
*
* @param string $run_id Run UUID.
* @param string $stage Stage name ('' for seeds).
* @param string $agent_role Agent role.
* @param string $item_key Item key.
* @param int $version Version number.
* @param string $source 'human' or 'seed'.
* @param string $request_json Payload JSON.
* @param string $author Author label.
* @return bool Whether the row was inserted.
*/
private static function insert_row( string $run_id, string $stage, string $agent_role, string $item_key, int $version, string $source, string $request_json, string $author ): bool {
global $wpdb;
// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery
$inserted = $wpdb->insert(
self::table_name(),
array(
'run_id' => $run_id,
'stage' => $stage,
'agent_role' => $agent_role,
'item_key' => $item_key,
'version' => $version,
'source' => $source,
'request_json' => $request_json,
'author' => $author,
'created_at' => current_time( 'mysql' ),
)
);
return false !== $inserted;
}
}