feat: expose persistent goal-loop state and completion controls via HTTP API

Add five new endpoints to the runtime API for managing thread goals:

  GET    /v1/threads/{id}/goal         — read goal (objective, status, usage,
                                         budget, continuation count, timestamps)
  PUT    /v1/threads/{id}/goal         — create or update goal (objective +
                                         optional token_budget)
  DELETE /v1/threads/{id}/goal         — clear goal, emits cleared event
  POST   /v1/threads/{id}/goal/complete — transition to complete (409 if already
                                          terminal)
  POST   /v1/threads/{id}/goal/block   — transition to blocked  (409 if already
                                          complete)

Goals are stored durably in the RuntimeThreadStore (goals/ subdirectory, one
JSON file per thread). Every write emits a replayable SSE event
(thread_goal_updated / thread_goal_cleared) so subscribers get the same
durable update that engine-driven changes produce.

The GET/PUT/DELETE handlers verify the thread exists in the runtime store
before touching the goal; unknown-thread requests return 404.

Lifecycle authority is preserved: PUT always resets to Active status; the
complete and block actions are the only paths to those states.

  • Add goals_dir + save_goal/load_goal/delete_goal to RuntimeThreadStore
  • Add get_goal/save_goal/remove_goal + goal event helpers to
    RuntimeThreadManager
  • Add thread_goals: bool to RuntimeCapabilities in codewhale-protocol
    (default false for old deserializers; set to true in runtime_api)
  • Four new integration tests covering CRUD, invalid transitions, auth,
    and capability advertisement

Closes #5074

crates/protocol/src/runtime/mod.rs

@@ -59,6 +59,10 @@ pub struct RuntimeCapabilities {
     pub fleet_event_stream: bool,
     #[serde(default)]
     pub fleet_local_target: bool,
+    /// `GET/PUT/DELETE /v1/threads/{id}/goal` and the `complete`/`block`
+    /// lifecycle actions are available.
+    #[serde(default)]
+    pub thread_goals: bool,
 }
 
 /// Experimental opt-in flags advertised by `GET /v1/runtime/info`.
@@ -355,6 +359,7 @@ mod tests {
             fleet_event_replay: true,
             fleet_event_stream: true,
             fleet_local_target: true,
+            thread_goals: true,
         };
         let value = serde_json::to_value(&caps).unwrap();
         let obj = value.as_object().unwrap();
@@ -364,6 +369,7 @@ mod tests {
         assert!(obj.contains_key("worker_runtime"));
         assert_eq!(obj.get("fleet_run_create").unwrap(), &json!(true));
         assert_eq!(obj.get("fleet_event_stream").unwrap(), &json!(true));
+        assert_eq!(obj.get("thread_goals").unwrap(), &json!(true));
     }
 
     #[test]

crates/tui/src/runtime_api.rs

@@ -401,6 +401,7 @@ fn default_runtime_capabilities() -> RuntimeCapabilities {
         fleet_event_replay: true,
         fleet_event_stream: true,
         fleet_local_target: true,
+        thread_goals: true,
     }
 }
 
@@ -751,6 +752,12 @@ pub fn build_router(state: RuntimeApiState) -> Router {
         )
         .route("/v1/threads/{id}/compact", post(compact_thread))
         .route("/v1/threads/{id}/events", get(stream_thread_events))
+        .route(
+            "/v1/threads/{id}/goal",
+            get(get_thread_goal).put(upsert_thread_goal).delete(delete_thread_goal),
+        )
+        .route("/v1/threads/{id}/goal/complete", post(complete_thread_goal))
+        .route("/v1/threads/{id}/goal/block", post(block_thread_goal))
         .route("/v1/approvals/{approval_id}", post(decide_approval))
         .route(
             "/v1/user-input/{thread_id}/{input_id}",
@@ -2801,6 +2808,214 @@ async fn compact_thread(
     ))
 }
 
+// ---------------------------------------------------------------------------
+// Thread goal endpoints
+// ---------------------------------------------------------------------------
+
+/// `GET /v1/threads/{id}/goal` — return the persistent goal for a thread, or
+/// 404 if the thread has no goal.
+async fn get_thread_goal(
+    State(state): State<RuntimeApiState>,
+    Path(id): Path<String>,
+) -> Result<Json<codewhale_protocol::ThreadGoal>, ApiError> {
+    // Verify the thread exists so we can return a clean 404 for unknown threads.
+    state
+        .runtime_threads
+        .get_thread(&id)
+        .await
+        .map_err(map_thread_err)?;
+    let goal = state
+        .runtime_threads
+        .get_goal(&id)
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?
+        .ok_or_else(|| ApiError::not_found(format!("thread '{id}' has no goal")))?;
+    Ok(Json(goal))
+}
+
+#[derive(Debug, Deserialize)]
+struct UpsertThreadGoalRequest {
+    objective: String,
+    #[serde(default)]
+    token_budget: Option<i64>,
+}
+
+/// `PUT /v1/threads/{id}/goal` — create or replace the persistent goal for a
+/// thread. Only `Active` goals may be created through this route; lifecycle
+/// transitions (`complete`, `block`) have dedicated action endpoints.
+async fn upsert_thread_goal(
+    State(state): State<RuntimeApiState>,
+    Path(id): Path<String>,
+    Json(req): Json<UpsertThreadGoalRequest>,
+) -> Result<(StatusCode, Json<codewhale_protocol::ThreadGoal>), ApiError> {
+    if req.objective.trim().is_empty() {
+        return Err(ApiError::bad_request("objective must not be blank"));
+    }
+    // Verify the thread exists.
+    state
+        .runtime_threads
+        .get_thread(&id)
+        .await
+        .map_err(map_thread_err)?;
+    let now = chrono::Utc::now().timestamp();
+    let existing = state
+        .runtime_threads
+        .get_goal(&id)
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?;
+    let is_new = existing.is_none();
+    let goal_id = existing
+        .as_ref()
+        .map(|g| g.goal_id.clone())
+        .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
+    let created_at = existing.as_ref().map(|g| g.created_at).unwrap_or(now);
+    let goal = codewhale_protocol::ThreadGoal {
+        thread_id: id.clone(),
+        goal_id,
+        objective: req.objective.clone(),
+        status: codewhale_protocol::ThreadGoalStatus::Active,
+        token_budget: req.token_budget,
+        tokens_used: existing.as_ref().map(|g| g.tokens_used).unwrap_or(0),
+        time_used_seconds: existing
+            .as_ref()
+            .map(|g| g.time_used_seconds)
+            .unwrap_or(0),
+        continuation_count: existing
+            .as_ref()
+            .map(|g| g.continuation_count)
+            .unwrap_or(0),
+        created_at,
+        updated_at: now,
+    };
+    state
+        .runtime_threads
+        .save_goal(goal.clone())
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?;
+    let status_code = if is_new {
+        StatusCode::CREATED
+    } else {
+        StatusCode::OK
+    };
+    // Emit a replayable goal-updated event so SSE subscribers can react.
+    let _ = state
+        .runtime_threads
+        .emit_goal_updated_event(&id, goal.clone())
+        .await;
+    Ok((status_code, Json(goal)))
+}
+
+/// `DELETE /v1/threads/{id}/goal` — remove the persistent goal from a thread.
+/// Returns 204 No Content on success, 404 if there was no goal.
+async fn delete_thread_goal(
+    State(state): State<RuntimeApiState>,
+    Path(id): Path<String>,
+) -> Result<StatusCode, ApiError> {
+    state
+        .runtime_threads
+        .get_thread(&id)
+        .await
+        .map_err(map_thread_err)?;
+    let deleted = state
+        .runtime_threads
+        .remove_goal(&id)
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?;
+    if !deleted {
+        return Err(ApiError::not_found(format!("thread '{id}' has no goal")));
+    }
+    let _ = state
+        .runtime_threads
+        .emit_goal_cleared_event(&id)
+        .await;
+    Ok(StatusCode::NO_CONTENT)
+}
+
+/// `POST /v1/threads/{id}/goal/complete` — transition the goal to `Complete`.
+/// Only valid from a non-terminal status; returns 409 Conflict if the goal is
+/// already in a terminal state, and 404 if the thread has no goal.
+async fn complete_thread_goal(
+    State(state): State<RuntimeApiState>,
+    Path(id): Path<String>,
+) -> Result<Json<codewhale_protocol::ThreadGoal>, ApiError> {
+    state
+        .runtime_threads
+        .get_thread(&id)
+        .await
+        .map_err(map_thread_err)?;
+    let goal = state
+        .runtime_threads
+        .get_goal(&id)
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?
+        .ok_or_else(|| ApiError::not_found(format!("thread '{id}' has no goal")))?;
+    if matches!(goal.status, codewhale_protocol::ThreadGoalStatus::Complete) {
+        return Err(ApiError {
+            status: StatusCode::CONFLICT,
+            message: format!("goal for thread '{id}' is already complete"),
+        });
+    }
+    let now = chrono::Utc::now().timestamp();
+    let updated = codewhale_protocol::ThreadGoal {
+        status: codewhale_protocol::ThreadGoalStatus::Complete,
+        updated_at: now,
+        ..goal
+    };
+    state
+        .runtime_threads
+        .save_goal(updated.clone())
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?;
+    let _ = state
+        .runtime_threads
+        .emit_goal_updated_event(&id, updated.clone())
+        .await;
+    Ok(Json(updated))
+}
+
+/// `POST /v1/threads/{id}/goal/block` — transition the goal to `Blocked`.
+/// Rejects transitions from terminal states (returns 409).
+async fn block_thread_goal(
+    State(state): State<RuntimeApiState>,
+    Path(id): Path<String>,
+) -> Result<Json<codewhale_protocol::ThreadGoal>, ApiError> {
+    state
+        .runtime_threads
+        .get_thread(&id)
+        .await
+        .map_err(map_thread_err)?;
+    let goal = state
+        .runtime_threads
+        .get_goal(&id)
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?
+        .ok_or_else(|| ApiError::not_found(format!("thread '{id}' has no goal")))?;
+    if matches!(goal.status, codewhale_protocol::ThreadGoalStatus::Complete) {
+        return Err(ApiError {
+            status: StatusCode::CONFLICT,
+            message: format!(
+                "goal for thread '{id}' is already complete; cannot transition to blocked"
+            ),
+        });
+    }
+    let now = chrono::Utc::now().timestamp();
+    let updated = codewhale_protocol::ThreadGoal {
+        status: codewhale_protocol::ThreadGoalStatus::Blocked,
+        updated_at: now,
+        ..goal
+    };
+    state
+        .runtime_threads
+        .save_goal(updated.clone())
+        .await
+        .map_err(|e| ApiError::internal(e.to_string()))?;
+    let _ = state
+        .runtime_threads
+        .emit_goal_updated_event(&id, updated.clone())
+        .await;
+    Ok(Json(updated))
+}
+
 async fn list_tasks(
     State(state): State<RuntimeApiState>,
     Query(query): Query<TasksQuery>,

crates/tui/src/runtime_api/tests.rs

@@ -7424,3 +7424,235 @@ async fn cors_layer_advertises_exact_supported_headers_and_never_an_extra() -> R
     handle.abort();
     Ok(())
 }
+
+// ---------------------------------------------------------------------------
+// Goal-loop endpoint tests
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+async fn thread_goal_crud_and_invalid_transition() -> Result<()> {
+    let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+
+    // Create a thread to associate goals with.
+    let thread: serde_json::Value = client
+        .post(format!("http://{addr}/v1/threads"))
+        .json(&json!({}))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    let thread_id = thread["id"].as_str().expect("thread id").to_string();
+
+    // GET before any goal exists → 404.
+    let no_goal = client
+        .get(format!("http://{addr}/v1/threads/{thread_id}/goal"))
+        .send()
+        .await?;
+    assert_eq!(no_goal.status(), 404, "no goal yet");
+
+    // PUT creates the goal (201).
+    let created: serde_json::Value = client
+        .put(format!("http://{addr}/v1/threads/{thread_id}/goal"))
+        .json(&json!({"objective": "write tests", "token_budget": 5000}))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(
+        created["status"].as_str(),
+        Some("active"),
+        "new goal is active"
+    );
+    assert_eq!(
+        created["objective"].as_str(),
+        Some("write tests"),
+        "objective round-trips"
+    );
+    assert_eq!(created["token_budget"], 5000, "budget round-trips");
+
+    // GET after create → 200 with the same fields.
+    let fetched: serde_json::Value = client
+        .get(format!("http://{addr}/v1/threads/{thread_id}/goal"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(fetched["objective"], created["objective"]);
+    assert_eq!(fetched["goal_id"], created["goal_id"]);
+
+    // PUT again (update) → 200.
+    let updated: serde_json::Value = client
+        .put(format!("http://{addr}/v1/threads/{thread_id}/goal"))
+        .json(&json!({"objective": "write even better tests"}))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(
+        updated["objective"].as_str(),
+        Some("write even better tests")
+    );
+    // goal_id is preserved across updates.
+    assert_eq!(updated["goal_id"], created["goal_id"]);
+    // After PUT re-create, status is reset to active.
+    assert_eq!(updated["status"].as_str(), Some("active"));
+
+    // POST /complete → 200 with status = complete.
+    let completed: serde_json::Value = client
+        .post(format!(
+            "http://{addr}/v1/threads/{thread_id}/goal/complete"
+        ))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(completed["status"].as_str(), Some("complete"));
+
+    // POST /complete again on a terminal goal → 409 Conflict.
+    let double_complete = client
+        .post(format!(
+            "http://{addr}/v1/threads/{thread_id}/goal/complete"
+        ))
+        .send()
+        .await?;
+    assert_eq!(
+        double_complete.status(),
+        409,
+        "completing an already-complete goal must be 409"
+    );
+
+    // POST /block on a terminal goal → 409 Conflict.
+    let block_after_complete = client
+        .post(format!("http://{addr}/v1/threads/{thread_id}/goal/block"))
+        .send()
+        .await?;
+    assert_eq!(
+        block_after_complete.status(),
+        409,
+        "blocking a complete goal must be 409"
+    );
+
+    // DELETE → 204.
+    let deleted = client
+        .delete(format!("http://{addr}/v1/threads/{thread_id}/goal"))
+        .send()
+        .await?;
+    assert_eq!(deleted.status(), 204, "delete returns 204");
+
+    // DELETE again → 404.
+    let double_delete = client
+        .delete(format!("http://{addr}/v1/threads/{thread_id}/goal"))
+        .send()
+        .await?;
+    assert_eq!(double_delete.status(), 404, "second delete is 404");
+
+    handle.abort();
+    Ok(())
+}
+
+#[tokio::test]
+async fn thread_goal_block_transition() -> Result<()> {
+    let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+
+    let thread: serde_json::Value = client
+        .post(format!("http://{addr}/v1/threads"))
+        .json(&json!({}))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    let thread_id = thread["id"].as_str().expect("thread id").to_string();
+
+    // Create an active goal.
+    client
+        .put(format!("http://{addr}/v1/threads/{thread_id}/goal"))
+        .json(&json!({"objective": "block me"}))
+        .send()
+        .await?
+        .error_for_status()?;
+
+    // Block it.
+    let blocked: serde_json::Value = client
+        .post(format!("http://{addr}/v1/threads/{thread_id}/goal/block"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(blocked["status"].as_str(), Some("blocked"));
+
+    // Can still complete from blocked state.
+    let completed: serde_json::Value = client
+        .post(format!(
+            "http://{addr}/v1/threads/{thread_id}/goal/complete"
+        ))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(completed["status"].as_str(), Some("complete"));
+
+    handle.abort();
+    Ok(())
+}
+
+#[tokio::test]
+async fn thread_goal_on_unknown_thread_returns_404() -> Result<()> {
+    let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+
+    let resp = client
+        .get(format!("http://{addr}/v1/threads/nonexistent-id/goal"))
+        .send()
+        .await?;
+    assert_eq!(resp.status(), 404);
+
+    let put_resp = client
+        .put(format!("http://{addr}/v1/threads/nonexistent-id/goal"))
+        .json(&json!({"objective": "ghost"}))
+        .send()
+        .await?;
+    assert_eq!(put_resp.status(), 404);
+
+    handle.abort();
+    Ok(())
+}
+
+#[tokio::test]
+async fn runtime_info_advertises_thread_goals_capability() -> Result<()> {
+    let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+
+    let info: serde_json::Value = client
+        .get(format!("http://{addr}/v1/runtime/info"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(
+        info["capabilities"]["thread_goals"].as_bool(),
+        Some(true),
+        "runtime info must advertise thread_goals capability"
+    );
+
+    handle.abort();
+    Ok(())
+}

crates/tui/src/runtime_threads.rs

@@ -792,6 +792,7 @@ pub struct RuntimeThreadStore {
     turns_dir: PathBuf,
     items_dir: PathBuf,
     events_dir: PathBuf,
+    goals_dir: PathBuf,
     state_path: PathBuf,
     event_lock_path: PathBuf,
     /// Serializes load-modify-save operations on thread records. The guard is
@@ -811,16 +812,19 @@ impl RuntimeThreadStore {
         let turns_dir = root.join("turns");
         let items_dir = root.join("items");
         let events_dir = root.join("events");
+        let goals_dir = root.join("goals");
         ensure_runtime_store_dir(&threads_dir)?;
         ensure_runtime_store_dir(&turns_dir)?;
         ensure_runtime_store_dir(&items_dir)?;
         ensure_runtime_store_dir(&events_dir)?;
+        ensure_runtime_store_dir(&goals_dir)?;
         let state_path = root.join("state.json");
         let store = Self {
             threads_dir,
             turns_dir,
             items_dir,
             events_dir,
+            goals_dir,
             state_path,
             event_lock_path: root.join(EVENT_TRANSACTION_LOCK_FILE),
             thread_mutation: Arc::new(parking_lot::Mutex::new(())),
@@ -900,6 +904,41 @@ impl RuntimeThreadStore {
         Self::record_path(&self.events_dir, thread_id, "jsonl", "thread id")
     }
 
+    fn goal_path(&self, thread_id: &str) -> Result<PathBuf> {
+        Self::record_path(&self.goals_dir, thread_id, "json", "thread id")
+    }
+
+    /// Persist a goal record for a thread. The goal is stored as a JSON file
+    /// in the `goals/` subdirectory; it is independent of the TUI state store
+    /// and requires only that the runtime thread exists.
+    pub fn save_goal(&self, goal: &codewhale_protocol::ThreadGoal) -> Result<()> {
+        write_json_atomic(&self.goal_path(&goal.thread_id)?, goal)
+    }
+
+    /// Load the goal for a thread, returning `None` if no goal has been set.
+    pub fn load_goal(&self, thread_id: &str) -> Result<Option<codewhale_protocol::ThreadGoal>> {
+        let path = self.goal_path(thread_id)?;
+        if !path.exists() {
+            return Ok(None);
+        }
+        let raw = read_store_file(&path)
+            .with_context(|| format!("Failed to read goal {}", path.display()))?;
+        let goal: codewhale_protocol::ThreadGoal = serde_json::from_str(&raw)
+            .with_context(|| format!("Failed to parse goal {}", path.display()))?;
+        Ok(Some(goal))
+    }
+
+    /// Remove the goal for a thread, returning `true` if one existed.
+    pub fn delete_goal(&self, thread_id: &str) -> Result<bool> {
+        let path = self.goal_path(thread_id)?;
+        if !path.exists() {
+            return Ok(false);
+        }
+        fs::remove_file(&path)
+            .with_context(|| format!("Failed to delete goal {}", path.display()))?;
+        Ok(true)
+    }
+
     pub fn save_thread(&self, thread: &ThreadRecord) -> Result<()> {
         write_json_atomic(&self.thread_path(&thread.id)?, thread)
     }
@@ -3051,6 +3090,60 @@ impl RuntimeThreadManager {
         self.event_tx.subscribe()
     }
 
+    /// Emit a durable `thread_goal_updated` event for the given thread.
+    ///
+    /// Called by the Runtime API goal handlers so SSE subscribers receive
+    /// goal lifecycle changes just like engine-driven updates.
+    pub async fn emit_goal_updated_event(
+        &self,
+        thread_id: &str,
+        goal: codewhale_protocol::ThreadGoal,
+    ) -> Result<RuntimeEventRecord> {
+        let payload = serde_json::json!({
+            "kind": "thread_goal_updated",
+            "goal": serde_json::to_value(&goal)
+                .unwrap_or(serde_json::Value::Null),
+        });
+        self.emit_event(thread_id, None, None, "thread_goal_updated", payload)
+            .await
+    }
+
+    /// Emit a durable `thread_goal_cleared` event for the given thread.
+    pub async fn emit_goal_cleared_event(&self, thread_id: &str) -> Result<RuntimeEventRecord> {
+        let payload = serde_json::json!({
+            "kind": "thread_goal_cleared",
+            "thread_id": thread_id,
+        });
+        self.emit_event(thread_id, None, None, "thread_goal_cleared", payload)
+            .await
+    }
+
+    /// Return the persistent goal for a thread, or `Ok(None)` if none exists.
+    pub async fn get_goal(&self, thread_id: &str) -> Result<Option<codewhale_protocol::ThreadGoal>> {
+        let thread_id = thread_id.to_string();
+        let store = self.store.clone();
+        tokio::task::spawn_blocking(move || store.load_goal(&thread_id))
+            .await
+            .context("goal load task panicked")?
+    }
+
+    /// Persist (create or replace) the goal for a thread.
+    pub async fn save_goal(&self, goal: codewhale_protocol::ThreadGoal) -> Result<()> {
+        let store = self.store.clone();
+        tokio::task::spawn_blocking(move || store.save_goal(&goal))
+            .await
+            .context("goal save task panicked")?
+    }
+
+    /// Remove the goal for a thread. Returns `true` if a goal existed.
+    pub async fn remove_goal(&self, thread_id: &str) -> Result<bool> {
+        let thread_id = thread_id.to_string();
+        let store = self.store.clone();
+        tokio::task::spawn_blocking(move || store.delete_goal(&thread_id))
+            .await
+            .context("goal delete task panicked")?
+    }
+
     fn projection_lock(&self, thread_id: &str) -> Arc<Mutex<()>> {
         let mut locks = self.projection_locks.lock();
         Arc::clone(