batch = $batch ?? new PRAutoBlogger_Research_Batch(); } /** * Dispatch N specialist research agents in parallel and collect results. * * Returns an indexed array of per-agent result maps. If fewer than * quorum (⌈N/2⌉+1) agents return usable results, returns an empty * array — the caller should HOLD the run rather than proceeding on * thin research. * * Side effects: curl_multi HTTP calls to OpenRouter, run_stages DB * writes, cost-governor reservation + settlement, Logger calls. * * @param string $run_id Run UUID. * @param string $item_key Article-scoped item key. * @param PRAutoBlogger_Article_Idea $idea The idea being researched. * @param PRAutoBlogger_Cost_Tracker $cost_tracker Pipeline cost tracker. * @return array, agent_role: string}> Per-agent results, or empty when quorum not met. */ public function dispatch( string $run_id, string $item_key, PRAutoBlogger_Article_Idea $idea, PRAutoBlogger_Cost_Tracker $cost_tracker ): array { $n = $this->resolve_agent_count(); $model = $this->resolve_model(); $quorum = (int) ceil( $n / 2 ) + 1; $roles = array_slice( self::SPECIALIST_ROLES, 0, $n ); $topic = $idea->get_topic(); $title = $idea->get_suggested_title(); $messages_per_agent = $this->build_agent_messages( $topic, $title, $roles ); $options = array( 'temperature' => 0.3, 'max_tokens' => 3000, 'response_format' => array( 'type' => 'json_object' ), ); // Reserve the SUMMED worst-case cost of all N agents before dispatch. $per_estimate = PRAutoBlogger_Cost_Governor::estimate_chat_cost( $model, $messages_per_agent[0], $options ); $batch_estimate = $per_estimate * $n; $reservation = PRAutoBlogger_Cost_Governor::open_amount_reservation( $batch_estimate, sprintf( 'research_fanout:n=%d:%s', $n, $model ) ); PRAutoBlogger_Logger::instance()->info( sprintf( 'Research fan-out: dispatching %d agents (quorum=%d) for "%s". Batch reserve: $%.4f.', $n, $quorum, mb_substr( $topic, 0, 60 ), $batch_estimate ), 'research-fanout' ); foreach ( $roles as $role ) { PRAutoBlogger_Run_Stage_State::start( $run_id, 'research', 'researcher:' . $role, $item_key ); } $valid_results = array(); $actual_total = 0.0; // try/finally: settle() ALWAYS runs — even if batch->execute() throws. // Without this, an uncaught exception leaves reserved_usd open in the ledger // until the run reaper, silently eroding the ceiling. Settles 0.0 on throw. try { $raw_results = $this->batch->execute( $model, $messages_per_agent, $options, $roles ); foreach ( $raw_results as $idx => $raw ) { $role = $roles[ $idx ]; if ( isset( $raw['error'] ) ) { PRAutoBlogger_Logger::instance()->warning( sprintf( 'Research agent "%s" failed: %s', $role, $raw['error'] ), 'research-fanout' ); PRAutoBlogger_Run_Stage_State::fail( $run_id, 'research', 'researcher:' . $role, $item_key ); continue; } $parsed = PRAutoBlogger_Json_Extractor::decode( $raw['content'] ?? '' ); $sources = $this->validate_sources( $parsed ); if ( null === $sources ) { PRAutoBlogger_Logger::instance()->warning( sprintf( 'Research agent "%s" returned invalid schema; excluding.', $role ), 'research-fanout' ); PRAutoBlogger_Run_Stage_State::fail( $run_id, 'research', 'researcher:' . $role, $item_key ); continue; } $actual_cost = (float) ( $raw['actual_cost'] ?? 0.0 ); $actual_total += $actual_cost; $cost_tracker->log_api_call( null, 'research', 'openrouter', $raw['model'] ?? $model, (int) ( $raw['prompt_tokens'] ?? 0 ), (int) ( $raw['completion_tokens'] ?? 0 ) ); PRAutoBlogger_Run_Stage_State::done( $run_id, 'research', 'researcher:' . $role, $item_key, wp_json_encode( array( 'source_count' => count( $sources ) ) ), $actual_cost ); $valid_results[] = array( 'sources' => $sources, 'agent_role' => 'researcher:' . $role, ); } } finally { PRAutoBlogger_Cost_Governor::settle( $reservation, $actual_total ); // Always settles (0 on throw). } $valid_count = count( $valid_results ); PRAutoBlogger_Logger::instance()->info( sprintf( 'Research fan-out complete: %d/%d agents usable (quorum=%d).', $valid_count, $n, $quorum ), 'research-fanout' ); if ( $valid_count < $quorum ) { PRAutoBlogger_Logger::instance()->warning( sprintf( 'Research quorum not met (%d/%d). Run should hold.', $valid_count, $quorum ), 'research-fanout' ); return array(); } return $valid_results; } // ── Private helpers ───────────────────────────────────────────────── /** * Read the configured agent count, clamped to [MIN_AGENTS, MAX_AGENTS]. * MIN_AGENTS=2 since quorum=⌈N/2⌉+1=2 for N=1 is unsatisfiable; see CONTEXT.md §Phase 2b. * * @return int */ private function resolve_agent_count(): int { $n = (int) get_option( 'prautoblogger_research_agent_count', 3 ); return max( self::MIN_AGENTS, min( self::MAX_AGENTS, $n ) ); } /** * Read the configured research model. * * @return string */ private function resolve_model(): string { return (string) get_option( 'prautoblogger_research_model', PRAUTOBLOGGER_DEFAULT_ANALYSIS_MODEL ); } /** * Build the per-agent message sets (system + specialist user prompt). * * @param string $topic Topic text. * @param string $title Suggested article title. * @param string[] $roles Specialist role names. * @return array> */ private function build_agent_messages( string $topic, string $title, array $roles ): array { $system = PRAutoBlogger_Prompt_Registry::render( 'research.system' ); $sets = array(); foreach ( $roles as $role ) { $sets[] = array( array( 'role' => 'system', 'content' => $system, ), array( 'role' => 'user', 'content' => sprintf( "Topic: %s\nProposed title: %s\n\nSpecialist focus: %s\n\n" . "Find 3-6 high-quality sources for this topic from a '%s' perspective. " . 'Return JSON: {"sources":[{"url":"","title":"","excerpt":"","relevance":0.0}]}', $topic, $title, $role, $role ), ), ); } return $sets; } /** * Validate and normalise the parsed sources from one agent's response. * * Schema: `{sources: [{url, title, excerpt, relevance}]}`. Any entry * missing both url and title is silently skipped. Returns null when * the top-level schema is missing or sources is empty after filtering. * * @param array|null $parsed Decoded JSON from one agent. * @return array|null * * Side effects: none. */ private function validate_sources( ?array $parsed ): ?array { if ( ! is_array( $parsed ) || ! isset( $parsed['sources'] ) || ! is_array( $parsed['sources'] ) ) { return null; } $out = array(); foreach ( $parsed['sources'] as $s ) { if ( ! is_array( $s ) ) { continue; } $url = sanitize_url( (string) ( $s['url'] ?? '' ) ); $title = sanitize_text_field( (string) ( $s['title'] ?? '' ) ); $excerpt = sanitize_textarea_field( (string) ( $s['excerpt'] ?? '' ) ); $relevance = max( 0.0, min( 1.0, (float) ( $s['relevance'] ?? 0.0 ) ) ); if ( '' === $url && '' === $title ) { continue; } $out[] = array( 'url' => $url, 'title' => $title, 'excerpt' => $excerpt, 'relevance' => $relevance, ); } return count( $out ) > 0 ? $out : null; } }