feat(engine): consolidate action execution, remove reflection, add learning missions

Three major changes to the v2 engine:

1. **Consolidated action execution** — `handle_execute_action` in Rust is now
   the single source of truth for lease lookup, policy check, lease consumption,
   action execution, event emission, and ActionResult message recording. The
   Python orchestrator no longer duplicates event/message logic. This fixes the
   empty call_id bug (OpenAI HTTP 400) and the missing tool_calls on assistant
   messages (Codex "No tool call found" error).

2. **Removed reflection system** — Deleted the per-thread reflection pipeline
   (pipeline.rs, executor.rs), ThreadState::Reflecting, ThreadType::Reflection,
   enable_reflection config, and all 3 reflection event kinds. Learning is now
   handled entirely by event-driven missions that fire selectively.

3. **Three learning missions** replace reflection:
   - `self-improvement` — fires on trace issues (error diagnosis, prompt fixes)
   - `playbook-extraction` — fires on successful 5+ step threads (reusable procedures)
   - `conversation-insights` — fires every 5 threads per project (user preferences,
     domain knowledge, workflow patterns)

Additional fixes:
- llm_query()/llm_query_batched() always include system message (Codex compat)
- handle_llm_complete adds assistant message with structured action_calls for
  Tier 0 responses (prevents "No tool call found" errors)
- Gateway broadcasts without thread_id emit as Status events instead of being dropped
- Comprehensive tests for call_id propagation and trace analysis (17 new tests)

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

src/bridge/router.rs

@@ -201,12 +201,12 @@ pub async fn init_engine(agent: &Agent) -> Result<(), Error> {
     mission_manager.start_cron_ticker(agent.deps.owner_id.clone());
     mission_manager.start_event_listener(agent.deps.owner_id.clone());
 
-    // Ensure self-improvement mission exists for this project
+    // Ensure all learning missions exist for this project
     if let Err(e) = mission_manager
-        .ensure_self_improvement_mission(project_id)
+        .ensure_learning_missions(project_id)
         .await
     {
-        debug!("engine v2: failed to create self-improvement mission: {e}");
+        debug!("engine v2: failed to create learning missions: {e}");
     }
 
     // Wire mission manager into effect adapter for mission_* function calls
@@ -816,10 +816,7 @@ pub async fn handle_with_engine(
             content,
             state.default_project_id,
             &message.user_id,
-            ThreadConfig {
-                enable_reflection: true,
-                ..ThreadConfig::default()
-            },
+            ThreadConfig::default(),
         )
         .await
         .map_err(|e| engine_err("thread error", e))?;
@@ -1108,26 +1105,6 @@ async fn forward_event_to_channel(
                     .await;
             }
         }
-        EventKind::ReflectionStarted => {
-            let _ = channels
-                .send_status(
-                    channel_name,
-                    StatusUpdate::Thinking("Reflecting on execution...".into()),
-                    metadata,
-                )
-                .await;
-        }
-        EventKind::ReflectionComplete { docs_produced, .. } => {
-            let _ = channels
-                .send_status(
-                    channel_name,
-                    StatusUpdate::Thinking(format!(
-                        "Reflection complete — {docs_produced} insight(s) saved"
-                    )),
-                    metadata,
-                )
-                .await;
-        }
         _ => {}
     }
 }
@@ -1211,14 +1188,6 @@ fn thread_event_to_app_events(
             .into_iter()
             .collect()
         }
-        EventKind::ReflectionStarted => vec![AppEvent::Thinking {
-            message: "Reflecting on execution...".into(),
-            thread_id: Some(thread_id.into()),
-        }],
-        EventKind::ReflectionComplete { docs_produced, .. } => vec![AppEvent::Status {
-            message: format!("Reflection complete — {docs_produced} insight(s) saved"),
-            thread_id: Some(thread_id.into()),
-        }],
         EventKind::StateChanged { from, to, reason } => {
             vec![AppEvent::ThreadStateChanged {
             thread_id: thread_id.into(),

src/channels/web/mod.rs

@@ -534,22 +534,17 @@ impl Channel for GatewayChannel {
         user_id: &str,
         response: OutgoingResponse,
     ) -> Result<(), ChannelError> {
-        let thread_id = match response.thread_id {
-            Some(tid) => tid,
-            None => {
-                tracing::warn!(
-                    "Gateway broadcast with no thread_id — skipping (clients would drop it)"
-                );
-                return Ok(());
-            }
-        };
-        self.state.sse.broadcast_for_user(
-            user_id,
-            AppEvent::Response {
+        let event = match response.thread_id {
+            Some(thread_id) => AppEvent::Response {
                 content: response.content,
                 thread_id,
             },
-        );
+            None => AppEvent::Status {
+                message: response.content,
+                thread_id: None,
+            },
+        };
+        self.state.sse.broadcast_for_user(user_id, event);
         Ok(())
     }