diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c291599..7d931bc2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,14 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Changed +- `actual advisor` now streams its answer over Server-Sent Events instead of polling. Phase progress (fetching, interpreting, summarizing) renders to stderr while the answer is prepared, and the final answer prints to stdout. The previous client-side five-minute cap is gone, so a longer ADR-backed question runs to completion. + +### Added +- `actual advisor --json` prints the answer as JSON on stdout for scripting; phase progress stays on stderr so stdout remains clean. + ## [0.1.4] - 2026-03-31 ### Fixed diff --git a/README.md b/README.md index 181586d6..8e43d552 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,7 @@ you've written yourself. ``` actual adr-bot # analyze repo & write AI context files +actual advisor "" # ask the Advisor an org-scoped architecture question actual status # check output file state (managed markers, staleness) actual auth # verify authentication actual config show # view current configuration @@ -130,6 +131,27 @@ actual models # list known model names grouped by runner actual cache clear # clear local analysis and tailoring caches ``` +## Ask the Advisor + +`actual advisor` asks the Actual AI Advisor an org-scoped architecture +question and streams the answer back over Server-Sent Events. Progress steps +(fetching, interpreting, summarizing) print to stderr as the answer is +prepared; the final answer prints to stdout, so it stays pipeable. There is +no fixed time limit, so a longer ADR-backed question runs to completion. + +```bash +actual login # one-time sign-in +actual advisor "Should new services use gRPC or REST?" +``` + +Scope the question to one connected repository, or emit the answer as JSON +for scripting: + +```bash +actual advisor --repo "..." # scope to one repo (org-wide if omitted) +actual advisor --json "..." | jq .summary # machine-readable answer on stdout +``` + ## Configuration Config lives at `~/.actualai/actual/config.yaml` and is created automatically diff --git a/src/api/client.rs b/src/api/client.rs index 59571bf8..1713254f 100644 --- a/src/api/client.rs +++ b/src/api/client.rs @@ -1,16 +1,15 @@ use std::collections::HashMap; use std::time::Duration; -use reqwest::header::{HeaderMap, HeaderValue, ETAG, IF_NONE_MATCH, RETRY_AFTER, USER_AGENT}; +use reqwest::header::{HeaderMap, HeaderValue, USER_AGENT}; use crate::analysis::signals::CanonicalIR; use crate::analysis::static_analyzer::registry::all_framework_names; use crate::analysis::types::{FrameworkCategory, Language, RepoAnalysis}; use crate::api::types::{ - AdvisorPoll, AdvisorQueryRequest, AdvisorQueryStarted, AdvisorQueryStatus, ApiErrorResponse, - CanonicalIRFacet, CanonicalIRPayload, CategoriesResponse, FrameworksResponse, HealthResponse, - LanguagesResponse, MatchFramework, MatchOptions, MatchProject, MatchRequest, MatchResponse, - TelemetryRequest, + AdvisorStreamRequest, ApiErrorResponse, CanonicalIRFacet, CanonicalIRPayload, + CategoriesResponse, FrameworksResponse, HealthResponse, LanguagesResponse, MatchFramework, + MatchOptions, MatchProject, MatchRequest, MatchResponse, TelemetryRequest, }; use crate::config::types::Config; use crate::error::ActualError; @@ -22,6 +21,10 @@ const USER_AGENT_VALUE: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_ #[derive(Debug)] pub struct ActualApiClient { client: reqwest::Client, + /// A second client used only for the advisor SSE stream. It has no overall + /// request timeout (a streamed answer can take minutes); a per-read idle + /// timeout bounds inactivity instead. + stream_client: reqwest::Client, base_url: String, bearer_token: Option, } @@ -31,6 +34,11 @@ impl ActualApiClient { const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); /// Default connection establishment timeout (10 seconds). const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + /// Idle (between-reads) timeout for the SSE stream. There is deliberately no + /// overall request timeout on the stream — advisor answers can take minutes — + /// so this idle bound is the liveness backstop. Server heartbeat frames keep + /// it from firing during a long phase. + const STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(120); pub fn new(base_url: &str) -> Result { Self::new_with_timeout( @@ -61,15 +69,27 @@ impl ActualApiClient { default_headers.insert(USER_AGENT, HeaderValue::from_static(USER_AGENT_VALUE)); let client = reqwest::Client::builder() - .default_headers(default_headers) + .default_headers(default_headers.clone()) .https_only(!is_localhost) .timeout(timeout) .connect_timeout(connect_timeout) .build() .map_err(|e| ActualError::ApiError(format!("Failed to build HTTP client: {e}")))?; + // The streaming client shares the headers and connect bound but drops the + // overall request timeout, relying on the per-read idle timeout to bound + // inactivity so a minutes-long advisor stream is not cut off mid-answer. + let stream_client = reqwest::Client::builder() + .default_headers(default_headers) + .https_only(!is_localhost) + .connect_timeout(connect_timeout) + .read_timeout(Self::STREAM_IDLE_TIMEOUT) + .build() + .map_err(|e| ActualError::ApiError(format!("Failed to build streaming client: {e}")))?; + Ok(Self { client, + stream_client, base_url: base_url.trim_end_matches('/').to_string(), bearer_token: None, }) @@ -90,14 +110,19 @@ impl ActualApiClient { } } - /// Start an asynchronous advisor query. Returns the `query_id` to poll. - pub async fn start_advisor_query( + /// Open the advisor SSE stream. POSTs the query to `/v1/advisor/query/stream` + /// and returns the raw streaming response for the caller to consume frame by + /// frame. Terminal status codes are mapped here so the caller only ever sees + /// a 2xx streaming response or an error: 401 → `NotLoggedIn`, 403 → + /// `OrgMismatch`, other non-2xx → the parsed API error. + pub async fn stream_advisor_query( &self, - request: &AdvisorQueryRequest, - ) -> Result { - let url = format!("{}/v1/advisor/query", self.base_url); + request: &AdvisorStreamRequest, + ) -> Result { + let url = format!("{}/v1/advisor/query/stream", self.base_url); let response = self - .authed(self.client.post(&url).json(request)) + .authed(self.stream_client.post(&url).json(request)) + .header(reqwest::header::ACCEPT, "text/event-stream") .send() .await .map_err(|e| ActualError::ApiError(e.to_string()))?; @@ -111,70 +136,7 @@ impl ActualApiClient { if !status.is_success() { return Err(Self::map_error_response(status, response).await); } - response - .json::() - .await - .map_err(|e| ActualError::ApiError(e.to_string())) - } - - /// Poll an advisor query once. Pass the terminal `ETag` as `if_none_match` - /// to receive `304 Not Modified` once the result has settled. - pub async fn poll_advisor_query( - &self, - query_id: &str, - if_none_match: Option<&str>, - ) -> Result { - let url = format!("{}/v1/advisor/query/{query_id}", self.base_url); - let mut builder = self.authed(self.client.get(&url)); - if let Some(etag) = if_none_match { - builder = builder.header(IF_NONE_MATCH, etag); - } - let response = match builder.send().await { - Ok(r) => r, - // A transient network/fetch failure while polling — keep the loop - // alive (bounded by the caller's attempt cap) instead of aborting the - // query, mirroring the browser reference's keep-alive-on-fetch-error. - Err(_) => return Ok(AdvisorPoll::Retry { retry_after: None }), - }; - let status = response.status(); - if status == reqwest::StatusCode::NOT_MODIFIED { - return Ok(AdvisorPoll::NotModified); - } - if status == reqwest::StatusCode::UNAUTHORIZED { - return Err(ActualError::NotLoggedIn); - } - let retry_after = response - .headers() - .get(RETRY_AFTER) - .and_then(|v| v.to_str().ok()) - .and_then(|s| s.trim().parse::().ok()); - // A 5xx while polling is a transient infrastructure error (e.g. the - // status row could not be loaded) — the job is still running server-side, - // so retry rather than abort. A 404 (unknown query / no org access) stays - // terminal below. - if status.is_server_error() { - return Ok(AdvisorPoll::Retry { retry_after }); - } - if status == reqwest::StatusCode::FORBIDDEN { - return Err(Self::forbidden_org_error()); - } - if !status.is_success() { - return Err(Self::map_error_response(status, response).await); - } - let etag = response - .headers() - .get(ETAG) - .and_then(|v| v.to_str().ok()) - .map(|s| s.to_string()); - let body = response - .json::() - .await - .map_err(|e| ActualError::ApiError(e.to_string()))?; - Ok(AdvisorPoll::Update { - status: body, - retry_after, - etag, - }) + Ok(response) } async fn get(&self, path: &str) -> Result { @@ -420,7 +382,7 @@ mod tests { Framework, FrameworkCategory, Language, LanguageStat, Project, ProjectSelection, RepoAnalysis, WorkspaceType, }; - use crate::api::types::{AdvisorJobStatus, AdvisorSink, AdvisorSurface}; + use crate::api::types::AdvisorStreamRequest; /// Helper: create a minimal project with the given languages and frameworks. fn make_project( @@ -1534,70 +1496,78 @@ mod tests { assert!(request.projects[0].frameworks.is_empty()); } - // --- Advisor query client tests --- + // --- Advisor stream client tests --- - fn sample_advisor_request() -> AdvisorQueryRequest { - AdvisorQueryRequest::new( + fn sample_stream_request() -> AdvisorStreamRequest { + AdvisorStreamRequest::new( "11111111-1111-1111-1111-111111111111".to_string(), - None, "why no global mutable state?".to_string(), - AdvisorSurface::cli(), - AdvisorSink::None, None, ) } #[tokio::test] - async fn test_start_advisor_query_success_sends_bearer() { + async fn test_stream_advisor_query_success_sends_bearer_and_body() { let mut server = mockito::Server::new_async().await; + let sse = "data: {\"type\":\"phase\",\"data\":{\"phase\":\"fetching\"}}\n\n\ + data: {\"type\":\"final\",\"data\":{\"summary\":\"S\",\"interpreter\":{\"summary\":\"i\",\"related_adrs\":[]}}}\n\n"; let mock = server - .mock("POST", "/v1/advisor/query") + .mock("POST", "/v1/advisor/query/stream") .match_header("authorization", "Bearer tok-123") + .match_body(mockito::Matcher::PartialJsonString( + r#"{"orgId":"11111111-1111-1111-1111-111111111111","context":"why no global mutable state?"}"# + .to_string(), + )) .with_status(200) - .with_header("content-type", "application/json") - .with_body(r#"{"query_id":"q1","workflow_id":"wf1","status":"pending"}"#) + .with_header("content-type", "text/event-stream") + .with_body(sse) .create_async() .await; let client = ActualApiClient::new(&server.url()) .unwrap() .with_bearer("tok-123"); - let started = client - .start_advisor_query(&sample_advisor_request()) + let response = client + .stream_advisor_query(&sample_stream_request()) .await - .unwrap(); - assert_eq!(started.query_id, "q1"); - assert_eq!(started.workflow_id.as_deref(), Some("wf1")); - assert_eq!(started.status, AdvisorJobStatus::Pending); + .expect("stream opens"); + let body = response.text().await.unwrap(); + assert!(body.contains("\"type\":\"final\"")); mock.assert_async().await; } #[tokio::test] - async fn test_start_advisor_query_without_bearer_omits_header() { + async fn test_stream_advisor_query_forwards_repo_unique_id() { let mut server = mockito::Server::new_async().await; let mock = server - .mock("POST", "/v1/advisor/query") - .match_header("authorization", mockito::Matcher::Missing) + .mock("POST", "/v1/advisor/query/stream") + .match_body(mockito::Matcher::PartialJsonString( + r#"{"repo_unique_id":"33333333-3333-3333-3333-333333333333"}"#.to_string(), + )) .with_status(200) - .with_header("content-type", "application/json") - .with_body(r#"{"query_id":"q2","status":"pending"}"#) + .with_header("content-type", "text/event-stream") + .with_body("data: {\"type\":\"error\",\"error\":\"x\"}\n\n") .create_async() .await; - // No bearer → exercises the `authed` None branch; workflow_id default. - let client = ActualApiClient::new(&server.url()).unwrap(); - let started = client - .start_advisor_query(&sample_advisor_request()) + let client = ActualApiClient::new(&server.url()) + .unwrap() + .with_bearer("t"); + let req = AdvisorStreamRequest::new( + "o".to_string(), + "q".to_string(), + Some("33333333-3333-3333-3333-333333333333".to_string()), + ); + let _ = client + .stream_advisor_query(&req) .await - .unwrap(); - assert_eq!(started.query_id, "q2"); - assert_eq!(started.workflow_id, None); + .expect("stream opens"); mock.assert_async().await; } #[tokio::test] - async fn test_start_advisor_query_unauthorized_maps_to_not_logged_in() { + async fn test_stream_advisor_query_unauthorized_maps_to_not_logged_in() { let mut server = mockito::Server::new_async().await; let mock = server - .mock("POST", "/v1/advisor/query") + .mock("POST", "/v1/advisor/query/stream") .with_status(401) .with_body("unauthorized") .create_async() @@ -1606,7 +1576,7 @@ mod tests { .unwrap() .with_bearer("bad"); let err = client - .start_advisor_query(&sample_advisor_request()) + .stream_advisor_query(&sample_stream_request()) .await .unwrap_err(); assert!(matches!(err, ActualError::NotLoggedIn)); @@ -1614,228 +1584,91 @@ mod tests { } #[tokio::test] - async fn test_start_advisor_query_error_status() { + async fn test_stream_advisor_query_forbidden_maps_to_org_mismatch() { let mut server = mockito::Server::new_async().await; - let body = - r#"{"error":{"code":"BAD_REQUEST","message":"org_id must be a uuid","details":null}}"#; + // api-service fail-closes a cross-org token with 403 before streaming. let mock = server - .mock("POST", "/v1/advisor/query") - .with_status(400) + .mock("POST", "/v1/advisor/query/stream") + .with_status(403) .with_header("content-type", "application/json") - .with_body(body) + .with_body(r#"{"error":{"code":"FORBIDDEN","message":"cross-org","details":null}}"#) .create_async() .await; let client = ActualApiClient::new(&server.url()) .unwrap() .with_bearer("t"); let err = client - .start_advisor_query(&sample_advisor_request()) + .stream_advisor_query(&sample_stream_request()) .await .unwrap_err(); assert!( - matches!(err, ActualError::ApiResponseError { ref code, .. } if code == "BAD_REQUEST") + matches!(err, ActualError::OrgMismatch { .. }), + "expected OrgMismatch for 403, got {err:?}" ); mock.assert_async().await; } #[tokio::test] - async fn test_poll_advisor_query_pending_with_retry_after() { - let mut server = mockito::Server::new_async().await; - let mock = server - .mock("GET", "/v1/advisor/query/q1") - .match_header("authorization", "Bearer t") - .with_status(200) - .with_header("content-type", "application/json") - .with_header("retry-after", "2") - .with_body(r#"{"query_id":"q1","status":"running","result":null,"error":null}"#) - .create_async() - .await; - let client = ActualApiClient::new(&server.url()) - .unwrap() - .with_bearer("t"); - let poll = client.poll_advisor_query("q1", None).await.unwrap(); - match poll { - AdvisorPoll::Update { - status, - retry_after, - etag, - } => { - assert_eq!(status.status, AdvisorJobStatus::Running); - assert!(!status.status.is_terminal()); - assert_eq!(retry_after, Some(2)); - assert_eq!(etag, None); - assert!(status.result.is_none()); - } - other => panic!("expected Update, got {other:?}"), - } - mock.assert_async().await; - } - - #[tokio::test] - async fn test_poll_advisor_query_terminal_with_etag_and_result() { + async fn test_stream_advisor_query_error_status() { let mut server = mockito::Server::new_async().await; - let body = r#"{"query_id":"q1","status":"succeeded","result":{"summary":"S","interpreter":{"summary":"IS","related_adrs":[{"id":"a1","name":"n","title":"t","policy":"p","instructions":"i","scope":"sc","relevance_reason":"r","confidence":0.9}]}},"error":null}"#; + let body = + r#"{"error":{"code":"BAD_REQUEST","message":"orgId must be a uuid","details":null}}"#; let mock = server - .mock("GET", "/v1/advisor/query/q1") - .with_status(200) + .mock("POST", "/v1/advisor/query/stream") + .with_status(400) .with_header("content-type", "application/json") - .with_header("etag", "\"v1\"") .with_body(body) .create_async() .await; let client = ActualApiClient::new(&server.url()) .unwrap() .with_bearer("t"); - let poll = client.poll_advisor_query("q1", None).await.unwrap(); - match poll { - AdvisorPoll::Update { - status, - retry_after, - etag, - } => { - assert_eq!(status.status, AdvisorJobStatus::Succeeded); - assert_eq!(retry_after, None); - assert_eq!(etag.as_deref(), Some("\"v1\"")); - let out = status.result.expect("result present"); - assert_eq!(out.summary, "S"); - assert_eq!(out.interpreter.summary, "IS"); - assert_eq!(out.interpreter.related_adrs.len(), 1); - assert_eq!(out.interpreter.related_adrs[0].confidence, 0.9); - } - other => panic!("expected Update, got {other:?}"), - } - mock.assert_async().await; - } - - #[tokio::test] - async fn test_poll_advisor_query_not_modified() { - let mut server = mockito::Server::new_async().await; - let mock = server - .mock("GET", "/v1/advisor/query/q1") - .match_header("if-none-match", "\"v1\"") - .with_status(304) - .create_async() - .await; - let client = ActualApiClient::new(&server.url()) - .unwrap() - .with_bearer("t"); - let poll = client - .poll_advisor_query("q1", Some("\"v1\"")) + let err = client + .stream_advisor_query(&sample_stream_request()) .await - .unwrap(); - assert_eq!(poll, AdvisorPoll::NotModified); - mock.assert_async().await; - } - - #[tokio::test] - async fn test_poll_advisor_query_unauthorized_maps_to_not_logged_in() { - let mut server = mockito::Server::new_async().await; - let mock = server - .mock("GET", "/v1/advisor/query/q1") - .with_status(401) - .create_async() - .await; - let client = ActualApiClient::new(&server.url()) - .unwrap() - .with_bearer("bad"); - let err = client.poll_advisor_query("q1", None).await.unwrap_err(); - assert!(matches!(err, ActualError::NotLoggedIn)); - mock.assert_async().await; - } - - #[tokio::test] - async fn test_poll_advisor_query_5xx_is_retryable() { - let mut server = mockito::Server::new_async().await; - let mock = server - .mock("GET", "/v1/advisor/query/q1") - .with_status(500) - .with_body("boom") - .create_async() - .await; - let client = ActualApiClient::new(&server.url()) - .unwrap() - .with_bearer("t"); - // A 5xx mid-poll is a transient infra error → retry, not a fatal error. - let poll = client.poll_advisor_query("q1", None).await.unwrap(); - assert!(matches!(poll, AdvisorPoll::Retry { .. })); - mock.assert_async().await; - } - - #[tokio::test] - async fn test_poll_advisor_query_404_is_terminal() { - let mut server = mockito::Server::new_async().await; - let mock = server - .mock("GET", "/v1/advisor/query/q1") - .with_status(404) - .with_header("content-type", "application/json") - .with_body(r#"{"error":"not found"}"#) - .create_async() - .await; - let client = ActualApiClient::new(&server.url()) - .unwrap() - .with_bearer("t"); - // A 404 (unknown query / no org access) is terminal — surfaces as an - // error rather than being retried. - let err = client.poll_advisor_query("q1", None).await.unwrap_err(); - assert!(matches!(err, ActualError::ApiError(_))); + .unwrap_err(); + assert!( + matches!(err, ActualError::ApiResponseError { ref code, .. } if code == "BAD_REQUEST") + ); mock.assert_async().await; } #[tokio::test] - async fn test_poll_advisor_query_network_error_is_retryable() { + async fn test_stream_advisor_query_network_error() { // Unroutable address → the request can't complete (connection refused). - // A transient fetch failure mid-poll should retry, not abort the query. let client = ActualApiClient::new("http://127.0.0.1:1") .unwrap() .with_bearer("t"); - let poll = client.poll_advisor_query("q1", None).await.unwrap(); - assert!(matches!(poll, AdvisorPoll::Retry { .. })); - } - - #[tokio::test] - async fn test_start_advisor_query_forbidden_maps_to_org_mismatch() { - let mut server = mockito::Server::new_async().await; - // api-service fail-closes a cross-org token with 403 on the start call. - let mock = server - .mock("POST", "/v1/advisor/query") - .with_status(403) - .with_header("content-type", "application/json") - .with_body(r#"{"error":{"code":"FORBIDDEN","message":"cross-org","details":null}}"#) - .create_async() - .await; - let client = ActualApiClient::new(&server.url()) - .unwrap() - .with_bearer("t"); let err = client - .start_advisor_query(&sample_advisor_request()) + .stream_advisor_query(&sample_stream_request()) .await .unwrap_err(); - assert!( - matches!(err, ActualError::OrgMismatch { .. }), - "expected OrgMismatch for 403, got {err:?}" - ); - mock.assert_async().await; + assert!(matches!(err, ActualError::ApiError(_))); } #[tokio::test] - async fn test_poll_advisor_query_forbidden_is_terminal_org_mismatch() { + async fn test_stream_advisor_query_without_bearer_omits_auth_header() { + // A client with no `with_bearer` takes the `None` arm of `authed`, so the + // request carries no Authorization header. `Matcher::Missing` fails the + // mock if one is sent, so this both opens the stream and asserts the omission. let mut server = mockito::Server::new_async().await; let mock = server - .mock("GET", "/v1/advisor/query/q1") - .with_status(403) - .with_header("content-type", "application/json") - .with_body(r#"{"error":{"code":"FORBIDDEN","message":"cross-org","details":null}}"#) + .mock("POST", "/v1/advisor/query/stream") + .match_header("authorization", mockito::Matcher::Missing) + .with_status(200) + .with_header("content-type", "text/event-stream") + .with_body( + "data: {\"type\":\"final\",\"data\":{\"summary\":\"S\",\"interpreter\":{\"summary\":\"i\",\"related_adrs\":[]}}}\n\n", + ) .create_async() .await; - let client = ActualApiClient::new(&server.url()) - .unwrap() - .with_bearer("t"); - // A 403 mid-poll is terminal (cross-org), NOT retried like a 5xx. - let err = client.poll_advisor_query("q1", None).await.unwrap_err(); - assert!( - matches!(err, ActualError::OrgMismatch { .. }), - "expected OrgMismatch for 403, got {err:?}" - ); + let client = ActualApiClient::new(&server.url()).unwrap(); + let response = client + .stream_advisor_query(&sample_stream_request()) + .await + .expect("stream opens without a bearer token"); + let body = response.text().await.unwrap(); + assert!(body.contains("\"type\":\"final\"")); mock.assert_async().await; } } diff --git a/src/api/types.rs b/src/api/types.rs index 8d01b526..dbc82b86 100644 --- a/src/api/types.rs +++ b/src/api/types.rs @@ -186,131 +186,37 @@ pub struct TelemetryMetric { pub tags: HashMap, } -// --- Advisor query types (POST /v1/advisor/query, GET /v1/advisor/query/:id) --- +// --- Advisor query types (POST /v1/advisor/query/stream) --- -/// The client surface initiating an advisor query. The CLI always sends `cli`. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub struct AdvisorSurface { - pub kind: String, -} - -impl AdvisorSurface { - /// The CLI surface (`{ "kind": "cli" }`). - pub fn cli() -> Self { - Self { - kind: "cli".to_string(), - } - } -} - -/// Result-delivery sink. The CLI polls for its own result, so it always sends -/// `none`; the other sink types exist server-side for the Slack/GitHub surfaces. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(tag = "type", rename_all = "snake_case")] -pub enum AdvisorSink { - None, -} - -/// Job-envelope discriminator. The server validates `type` as the exact literal -/// `advisor_query`, so this serializes to that string. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] -#[serde(rename_all = "snake_case")] -pub enum AdvisorJobType { - AdvisorQuery, -} - -/// Versioned `data` payload for an advisor-query job (server contract v1). -#[derive(Debug, Clone, PartialEq, Serialize)] -pub struct AdvisorQueryData { - pub query: String, -} - -/// Fixed job-data version the server validates as the literal `1`. -const ADVISOR_QUERY_VERSION_V1: u8 = 1; - -/// Request body for `POST /v1/advisor/query`. +/// Request body for `POST /v1/advisor/query/stream`. /// -/// The server expects a typed, versioned job envelope: `type`/`version` literals -/// plus the query nested under `data`. Build with [`AdvisorQueryRequest::new`] -/// so those invariants live in one place. +/// The streaming endpoint takes a flat body: `orgId` (camelCase, matching the +/// server's wire contract), the question as `context`, and an optional +/// `repo_unique_id` to scope the answer to one connected repo (org-wide when +/// omitted). #[derive(Debug, Clone, PartialEq, Serialize)] -pub struct AdvisorQueryRequest { +pub struct AdvisorStreamRequest { + #[serde(rename = "orgId")] pub org_id: String, + pub context: String, #[serde(skip_serializing_if = "Option::is_none")] pub repo_unique_id: Option, - #[serde(rename = "type")] - pub job_type: AdvisorJobType, - pub version: u8, - pub data: AdvisorQueryData, - pub surface: AdvisorSurface, - pub sink: AdvisorSink, - #[serde(skip_serializing_if = "Option::is_none")] - pub idempotency_key: Option, } -impl AdvisorQueryRequest { - /// Build a v1 advisor-query request. `type` and `version` are fixed by the - /// server's job-envelope contract; `query` is nested under `data`. - pub fn new( - org_id: String, - repo_unique_id: Option, - query: String, - surface: AdvisorSurface, - sink: AdvisorSink, - idempotency_key: Option, - ) -> Self { +impl AdvisorStreamRequest { + /// Build a stream request: `org_id` is the org to scope to, `context` is the + /// question, `repo_unique_id` narrows the answer to one connected repo. + pub fn new(org_id: String, context: String, repo_unique_id: Option) -> Self { Self { org_id, + context, repo_unique_id, - job_type: AdvisorJobType::AdvisorQuery, - version: ADVISOR_QUERY_VERSION_V1, - data: AdvisorQueryData { query }, - surface, - sink, - idempotency_key, } } } -/// Lifecycle state of an advisor job. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum AdvisorJobStatus { - Pending, - Running, - Succeeded, - Failed, -} - -impl AdvisorJobStatus { - /// `true` once the job has reached a terminal state. - pub fn is_terminal(self) -> bool { - matches!(self, Self::Succeeded | Self::Failed) - } -} - -/// Response from `POST /v1/advisor/query` — the job has been accepted. -#[derive(Debug, Clone, PartialEq, Deserialize)] -pub struct AdvisorQueryStarted { - pub query_id: String, - #[serde(default)] - pub workflow_id: Option, - pub status: AdvisorJobStatus, -} - -/// Status + result of an advisor job (`GET /v1/advisor/query/:id`). Extra -/// server fields (request echo, sink delivery, timestamps) are ignored. -#[derive(Debug, Clone, PartialEq, Deserialize)] -pub struct AdvisorQueryStatus { - pub query_id: String, - pub status: AdvisorJobStatus, - #[serde(default)] - pub result: Option, - #[serde(default)] - pub error: Option, -} - -/// The advisor's answer. +/// The advisor's answer — the `data` payload of the stream's `final` frame +/// (`{ summary, interpreter: { summary, related_adrs } }`). #[derive(Debug, Clone, PartialEq, Deserialize)] pub struct AdvisorOutput { pub summary: String, @@ -337,22 +243,6 @@ pub struct RelatedAdr { pub confidence: f64, } -/// Outcome of a single poll of `GET /v1/advisor/query/:id`. -#[derive(Debug, Clone, PartialEq)] -pub enum AdvisorPoll { - /// `304 Not Modified` — the terminal result is unchanged from the ETag sent. - NotModified, - /// A transient `5xx` (e.g. the status row could not be loaded due to an - /// infrastructure error) — retry the poll rather than failing the query. - Retry { retry_after: Option }, - /// A status body, with the server's `Retry-After` (seconds) and `ETag` hints. - Update { - status: AdvisorQueryStatus, - retry_after: Option, - etag: Option, - }, -} - #[cfg(test)] mod tests { use super::*; @@ -930,106 +820,65 @@ mod tests { assert_eq!(details["reason"], "empty array"); } - // --- Advisor query type tests --- - - #[test] - fn test_advisor_surface_cli() { - assert_eq!(AdvisorSurface::cli().kind, "cli"); - } - - #[test] - fn test_advisor_job_status_is_terminal() { - assert!(!AdvisorJobStatus::Pending.is_terminal()); - assert!(!AdvisorJobStatus::Running.is_terminal()); - assert!(AdvisorJobStatus::Succeeded.is_terminal()); - assert!(AdvisorJobStatus::Failed.is_terminal()); - } - - #[test] - fn test_advisor_job_status_serde() { - assert_eq!( - serde_json::to_value(AdvisorJobStatus::Succeeded).unwrap(), - serde_json::json!("succeeded") - ); - let s: AdvisorJobStatus = serde_json::from_value(serde_json::json!("failed")).unwrap(); - assert_eq!(s, AdvisorJobStatus::Failed); - } - - #[test] - fn test_advisor_sink_none_serializes_with_type_tag() { - assert_eq!( - serde_json::to_value(AdvisorSink::None).unwrap(), - serde_json::json!({ "type": "none" }) - ); - } + // --- Advisor stream type tests --- #[test] - fn test_advisor_query_request_serialization() { - let req = AdvisorQueryRequest::new( + fn test_advisor_stream_request_serialization_omits_absent_repo() { + let req = AdvisorStreamRequest::new( "11111111-1111-1111-1111-111111111111".to_string(), - None, - "why?".to_string(), - AdvisorSurface::cli(), - AdvisorSink::None, + "why app router?".to_string(), None, ); let v = serde_json::to_value(&req).unwrap(); - // Typed/versioned job envelope: type + version literals, query under data. - assert_eq!(v["type"], "advisor_query"); - assert_eq!(v["version"], 1); - assert_eq!(v["data"]["query"], "why?"); - assert!(v.get("query").is_none()); // not top-level anymore - assert_eq!(v["surface"]["kind"], "cli"); - assert_eq!(v["sink"]["type"], "none"); - assert_eq!(v["org_id"], "11111111-1111-1111-1111-111111111111"); - // Absent optionals are omitted. + // orgId is camelCase per the wire contract; the question rides on `context`. + assert_eq!(v["orgId"], "11111111-1111-1111-1111-111111111111"); + assert_eq!(v["context"], "why app router?"); + // Absent repo scope is omitted entirely (org-wide query). assert!(v.get("repo_unique_id").is_none()); - assert!(v.get("idempotency_key").is_none()); + // No leftover poll-envelope fields. + assert!(v.get("org_id").is_none()); + assert!(v.get("type").is_none()); + assert!(v.get("data").is_none()); } #[test] - fn test_advisor_query_request_includes_optionals_when_set() { - let req = AdvisorQueryRequest::new( + fn test_advisor_stream_request_includes_repo_when_set() { + let req = AdvisorStreamRequest::new( "o".to_string(), - Some("22222222-2222-2222-2222-222222222222".to_string()), "q".to_string(), - AdvisorSurface::cli(), - AdvisorSink::None, - Some("idem-1".to_string()), + Some("33333333-3333-3333-3333-333333333333".to_string()), ); let v = serde_json::to_value(&req).unwrap(); - assert_eq!(v["repo_unique_id"], "22222222-2222-2222-2222-222222222222"); - assert_eq!(v["idempotency_key"], "idem-1"); + assert_eq!(v["repo_unique_id"], "33333333-3333-3333-3333-333333333333"); } #[test] - fn test_advisor_query_status_deserialize_ignores_extra_fields() { - // The server echoes request/sink_delivery/timestamps; we ignore them. + fn test_advisor_output_deserialize_full_and_extra_fields_ignored() { + // The `final` frame's `data` is the full AdvisorOutput. Any extra server + // fields are ignored. let json = r#"{ - "query_id": "q1", - "workflow_id": "wf1", - "status": "succeeded", - "request": { "anything": true }, - "sink_delivery": null, - "created_at": "2026-05-29T00:00:00Z", - "result": { - "summary": "top-level summary", - "interpreter": { - "summary": "interp summary", - "related_adrs": [{ - "id": "a1", "name": "n1", "title": "t1", "policy": "p1", - "instructions": "i1", "scope": "s1", "relevance_reason": "r1", - "confidence": 0.75 - }] - } - }, - "error": null + "summary": "top-level summary", + "extra": "ignored", + "interpreter": { + "summary": "interp summary", + "related_adrs": [{ + "id": "a1", "name": "n1", "title": "t1", "policy": "p1", + "instructions": "i1", "scope": "s1", "relevance_reason": "r1", + "confidence": 0.75 + }] + } }"#; - let status: AdvisorQueryStatus = serde_json::from_str(json).unwrap(); - assert_eq!(status.status, AdvisorJobStatus::Succeeded); - let out = status.result.expect("result present"); + let out: AdvisorOutput = serde_json::from_str(json).unwrap(); assert_eq!(out.summary, "top-level summary"); + assert_eq!(out.interpreter.summary, "interp summary"); assert_eq!(out.interpreter.related_adrs.len(), 1); assert_eq!(out.interpreter.related_adrs[0].confidence, 0.75); } + + #[test] + fn test_advisor_output_deserialize_missing_related_adrs_defaults_empty() { + let json = r#"{"summary": "s", "interpreter": {"summary": "i"}}"#; + let out: AdvisorOutput = serde_json::from_str(json).unwrap(); + assert!(out.interpreter.related_adrs.is_empty()); + } } diff --git a/src/cli/args.rs b/src/cli/args.rs index 78d3d34e..8f9db021 100644 --- a/src/cli/args.rs +++ b/src/cli/args.rs @@ -365,6 +365,12 @@ pub struct AdvisorArgs { /// available yet — it requires the connected-repos API. #[arg(long, value_name = "REPO_ID")] pub repo: Option, + + /// Print the final answer as JSON on stdout instead of the human-readable + /// rendering. Phase progress still goes to stderr, so the JSON on stdout + /// stays clean and pipeable. + #[arg(long)] + pub json: bool, } /// Arguments for the `adr-bot` command diff --git a/src/cli/commands/advisor.rs b/src/cli/commands/advisor.rs index ce5bbaa3..9834ae07 100644 --- a/src/cli/commands/advisor.rs +++ b/src/cli/commands/advisor.rs @@ -1,17 +1,15 @@ //! `actual advisor ` — ask the Advisor org-scoped architecture questions. //! -//! Starts an async advisor job, polls to completion (honoring the server's -//! `Retry-After` and retrying transient network/5xx errors up to a ~5-minute -//! cap), and renders the answer. Uses the platform token from `actual login` -//! as the bearer. - -use std::time::{Duration, Instant}; +//! Opens an SSE stream (`POST /v1/advisor/query/stream`), renders phase progress +//! to stderr as it arrives, and prints the final answer to stdout. Uses the +//! platform token from `actual login` as the bearer. There is no client-side +//! time cap, so a long ADR-backed answer runs to completion; a per-read idle +//! timeout (kept alive by server heartbeats) is the only liveness bound. use chrono::{Duration as ChronoDuration, Utc}; +use futures_util::StreamExt; -use crate::api::types::{ - AdvisorJobStatus, AdvisorOutput, AdvisorPoll, AdvisorQueryRequest, AdvisorSink, AdvisorSurface, -}; +use crate::api::types::{AdvisorOutput, AdvisorStreamRequest}; use crate::api::{ActualApiClient, DEFAULT_API_URL}; use crate::auth::oauth; use crate::auth::store::{self, StoredCredentials}; @@ -19,23 +17,12 @@ use crate::cli::args::AdvisorArgs; use crate::cli::ui::theme; use crate::error::ActualError; -/// Hard wall-clock cap on polling before giving up, matching the browser -/// reference. A true deadline (not an attempt count) is what actually bounds -/// total time once the per-poll `Retry-After` back-off varies. -const HARD_TIMEOUT: Duration = Duration::from_secs(5 * 60); -/// Default delay between polls when the server provides no `Retry-After`. -const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(2); -/// Upper bound on a server-supplied `Retry-After`. The status handler caps its -/// own back-off at 15s, so a larger (or misbehaving) value must not let a single -/// poll stall the whole query. -const MAX_RETRY_AFTER: Duration = Duration::from_secs(15); - pub fn exec(args: &AdvisorArgs) -> Result<(), ActualError> { - build_runtime()?.block_on(run(args, HARD_TIMEOUT, DEFAULT_POLL_INTERVAL)) + build_runtime()?.block_on(run(args)) } /// Single-threaded tokio runtime so the sync CLI dispatch path can drive the -/// async query flow (mirrors `commands::login`). +/// async stream flow (mirrors `commands::login`). fn build_runtime() -> Result { tokio::runtime::Builder::new_current_thread() .enable_all() @@ -115,17 +102,7 @@ async fn ensure_fresh(creds: StoredCredentials) -> Result), - Failed(Option), -} - -async fn run( - args: &AdvisorArgs, - deadline: Duration, - poll_interval: Duration, -) -> Result<(), ActualError> { +async fn run(args: &AdvisorArgs) -> Result<(), ActualError> { let creds = store::load()?.ok_or(ActualError::NotLoggedIn)?; let creds = ensure_fresh(creds).await?; let org_id = args @@ -139,86 +116,166 @@ async fn run( let base_url = resolve_api_url(args.api_url.as_deref()); let client = ActualApiClient::new(&base_url)?.with_bearer(&creds.access_token); - let request = AdvisorQueryRequest::new( - org_id.clone(), - args.repo.clone(), - args.query.clone(), - AdvisorSurface::cli(), - AdvisorSink::None, - None, - ); - - let started = client - .start_advisor_query(&request) + let request = AdvisorStreamRequest::new(org_id.clone(), args.query.clone(), args.repo.clone()); + + let response = client + .stream_advisor_query(&request) .await .map_err(|e| enrich_org_mismatch(e, &session_org, &org_id, explicit_org))?; - eprintln!("{} thinking…", theme::hint("advisor")); - let outcome = poll_to_completion(&client, &started.query_id, deadline, poll_interval) + + let outcome = consume_stream(response) .await .map_err(|e| enrich_org_mismatch(e, &session_org, &org_id, explicit_org))?; match outcome { - Outcome::Succeeded(output) => { - print_answer(&output); + StreamOutcome::Final(data) => { + print_final(&data, args.json)?; Ok(()) } - Outcome::Failed(error) => Err(ActualError::ApiError(format!( - "Advisor query failed: {}", - error.unwrap_or_else(|| "unknown error".to_string()) - ))), + // The stream closed without a `final` frame and without an `error`. Per + // the wire contract a graceful close is completion, not a failure — there + // is simply no answer body to print. + StreamOutcome::Closed => Ok(()), } } -/// Poll the job until it reaches a terminal state, or the wall-clock `deadline` -/// elapses (a true time bound — an attempt count can't bound total time once the -/// server's `Retry-After` back-off varies). -async fn poll_to_completion( - client: &ActualApiClient, - query_id: &str, - deadline: Duration, - poll_interval: Duration, -) -> Result { - let start = Instant::now(); - while start.elapsed() < deadline { - match client.poll_advisor_query(query_id, None).await? { - AdvisorPoll::Update { - status, - retry_after, - .. - } => match status.status { - AdvisorJobStatus::Succeeded => { - return Ok(match status.result { - Some(output) => Outcome::Succeeded(Box::new(output)), - None => Outcome::Failed(Some("advisor returned no result".to_string())), - }); - } - AdvisorJobStatus::Failed => return Ok(Outcome::Failed(status.error)), - AdvisorJobStatus::Pending | AdvisorJobStatus::Running => { - sleep_for(retry_after, poll_interval).await; - } - }, - AdvisorPoll::NotModified => sleep_for(None, poll_interval).await, - // Transient infra 5xx — back off (honoring Retry-After) and re-poll. - AdvisorPoll::Retry { retry_after } => sleep_for(retry_after, poll_interval).await, +/// Terminal outcome of consuming the advisor stream. An `error` frame and a +/// transport failure are surfaced as `Err` by [`consume_stream`], so the only +/// success shapes are a `final` answer or a graceful close without one. +#[derive(Debug, PartialEq)] +enum StreamOutcome { + Final(serde_json::Value), + Closed, +} + +/// Drive the SSE response to a terminal outcome: render each `phase` frame to +/// stderr, return the `data` of the `final` frame, and surface an `error` frame +/// as a non-zero-exit `ApiError`. Frames split across read chunks are reassembled +/// by [`SseDecoder`]; a graceful close without a `final` frame is completion. +async fn consume_stream(response: reqwest::Response) -> Result { + let mut decoder = SseDecoder::default(); + let stream = response.bytes_stream(); + futures_util::pin_mut!(stream); + while let Some(chunk) = stream.next().await { + let bytes = + chunk.map_err(|e| ActualError::ApiError(format!("advisor stream read failed: {e}")))?; + for payload in decoder.push(bytes.as_ref()) { + if let Some(outcome) = handle_payload(&payload)? { + return Ok(outcome); + } } } - Err(ActualError::ApiError( - "Advisor query did not reach a result in time.".to_string(), - )) + // Stream ended. Flush a trailing event the server closed without a blank-line + // terminator, in case it carried the `final` frame. + if let Some(payload) = decoder.flush() { + if let Some(outcome) = handle_payload(&payload)? { + return Ok(outcome); + } + } + Ok(StreamOutcome::Closed) +} + +/// Interpret one decoded SSE payload. Renders phase progress to stderr as a side +/// effect; returns `Some(outcome)` once a terminal `final` frame arrives, `Err` +/// on an `error` frame, and `None` for frames that do not end the stream +/// (phase, heartbeat, unrecognized type, or a non-JSON line). +fn handle_payload(payload: &str) -> Result, ActualError> { + match parse_advisor_frame(payload) { + Some(AdvisorStreamEvent::Phase(phase)) => { + render_phase(&phase); + Ok(None) + } + Some(AdvisorStreamEvent::Final(data)) => Ok(Some(StreamOutcome::Final(data))), + Some(AdvisorStreamEvent::Error(message)) => Err(ActualError::ApiError(format!( + "Advisor query failed: {message}" + ))), + // Heartbeats, unrecognized frame types, and non-JSON lines are ignored. + Some(AdvisorStreamEvent::Heartbeat) | Some(AdvisorStreamEvent::Other(_)) | None => Ok(None), + } } -/// The next-poll delay: the server's `Retry-After` seconds, or the default -/// interval, **clamped to `MAX_RETRY_AFTER`** so a large or misbehaving -/// `Retry-After` can't stall a single poll past the wall-clock deadline's intent. -fn next_delay(retry_after: Option, default: Duration) -> Duration { - retry_after - .map(Duration::from_secs) - .unwrap_or(default) - .min(MAX_RETRY_AFTER) +/// A decoded SSE frame from the advisor stream. +#[derive(Debug, Clone, PartialEq)] +enum AdvisorStreamEvent { + /// `data.phase` progress (`fetching` | `interpreting` | `summarizing` | `done`). + Phase(String), + /// The terminal answer — the `data` payload (a serialized `AdvisorOutput`). + Final(serde_json::Value), + /// An `error` frame; its message ends the run with a non-zero exit. + Error(String), + /// A keep-alive; ignored. + Heartbeat, + /// A frame whose `type` is unrecognized; ignored. + Other(String), } -async fn sleep_for(retry_after: Option, default: Duration) { - tokio::time::sleep(next_delay(retry_after, default)).await; +/// Parse one SSE `data:` payload (the JSON after `data: `) into an event. +/// Returns `None` when the payload is not a JSON object with a string `type`, +/// so a stray or non-conforming line is ignored rather than fatal. +fn parse_advisor_frame(payload: &str) -> Option { + let value: serde_json::Value = serde_json::from_str(payload.trim()).ok()?; + let obj = value.as_object()?; + let kind = obj.get("type")?.as_str()?; + let event = match kind { + "phase" => { + let phase = obj + .get("data") + .and_then(|d| d.get("phase")) + .and_then(|p| p.as_str()) + .unwrap_or("") + .to_string(); + AdvisorStreamEvent::Phase(phase) + } + "final" => { + let data = obj.get("data").cloned().unwrap_or(serde_json::Value::Null); + AdvisorStreamEvent::Final(data) + } + "error" => { + let message = obj + .get("error") + .and_then(|e| e.as_str()) + .unwrap_or("unknown error") + .to_string(); + AdvisorStreamEvent::Error(message) + } + "heartbeat" => AdvisorStreamEvent::Heartbeat, + other => AdvisorStreamEvent::Other(other.to_string()), + }; + Some(event) +} + +/// Render one phase as a single stderr line. `summarizing` is a one-shot +/// deterministic step, so it renders as one line (not a typing animation); the +/// `done` marker is silent because the `final` frame carries the answer. +fn render_phase(phase: &str) { + let label = match phase { + "fetching" => "fetching ADRs…", + "interpreting" => "interpreting…", + "summarizing" => "summarizing…", + "" | "done" => return, + other => { + eprintln!("{} {other}…", theme::hint("advisor")); + return; + } + }; + eprintln!("{} {label}", theme::hint("advisor")); +} + +/// Print the final answer to stdout. With `--json`, echo the server's +/// `AdvisorOutput` object verbatim (pretty-printed) so machine consumers get the +/// canonical shape; otherwise render the human-readable form. +fn print_final(data: &serde_json::Value, json: bool) -> Result<(), ActualError> { + if json { + let rendered = serde_json::to_string_pretty(data) + .map_err(|e| ActualError::ApiError(format!("failed to render advisor JSON: {e}")))?; + println!("{rendered}"); + return Ok(()); + } + let output: AdvisorOutput = serde_json::from_value(data.clone()).map_err(|e| { + ActualError::ApiError(format!("advisor returned an unparseable answer: {e}")) + })?; + print_answer(&output); + Ok(()) } /// Render the advisor answer. **Never prints token material.** @@ -237,6 +294,79 @@ fn print_answer(output: &AdvisorOutput) { } } +/// Reassembles SSE frames from a byte stream. Bytes are buffered until a blank +/// line (`\n\n`, or the CRLF form `\n\r\n`) terminates an event; each complete +/// event's `data:` payload is then returned. A frame split across read chunks is +/// held in the buffer until its terminator arrives. +#[derive(Default)] +struct SseDecoder { + buf: Vec, +} + +impl SseDecoder { + /// Append a chunk and return the `data:` payload of every event that is now + /// complete. Events with no `data:` line (e.g. a comment-only keep-alive) + /// yield nothing. + fn push(&mut self, chunk: &[u8]) -> Vec { + self.buf.extend_from_slice(chunk); + let mut out = Vec::new(); + while let Some(end) = Self::event_boundary(&self.buf) { + let block: Vec = self.buf.drain(..end).collect(); + if let Some(payload) = Self::extract_data(&block) { + out.push(payload); + } + } + out + } + + /// Flush a trailing event the stream closed without a blank-line terminator. + /// Returns its `data:` payload if it has one. + fn flush(&mut self) -> Option { + if self.buf.is_empty() { + return None; + } + let block = std::mem::take(&mut self.buf); + Self::extract_data(&block) + } + + /// Index just past the first blank-line event terminator in `buf`, if any. + /// Recognizes both `\n\n` (LF) and `\n\r\n` (CRLF) blank lines. The boundary + /// always falls on an ASCII byte, so a drained block never splits a UTF-8 + /// codepoint. + fn event_boundary(buf: &[u8]) -> Option { + for i in 0..buf.len() { + if buf[i] == b'\n' { + if buf.get(i + 1) == Some(&b'\n') { + return Some(i + 2); + } + if buf.get(i + 1) == Some(&b'\r') && buf.get(i + 2) == Some(&b'\n') { + return Some(i + 3); + } + } + } + None + } + + /// Collect the `data:` lines of one event block, strip the `data:` prefix + /// (and one optional leading space), and join multiple data lines with `\n` + /// per the SSE spec. Returns `None` when the block has no data line. + fn extract_data(block: &[u8]) -> Option { + let text = String::from_utf8_lossy(block); + let mut data_lines = Vec::new(); + for raw_line in text.split('\n') { + let line = raw_line.strip_suffix('\r').unwrap_or(raw_line); + if let Some(rest) = line.strip_prefix("data:") { + data_lines.push(rest.strip_prefix(' ').unwrap_or(rest).to_string()); + } + } + if data_lines.is_empty() { + None + } else { + Some(data_lines.join("\n")) + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -244,8 +374,25 @@ mod tests { use crate::testutil::{EnvGuard, ENV_MUTEX}; use tempfile::tempdir; - const POLL_PATH: &str = "/v1/advisor/query/q1"; - const START_BODY: &str = r#"{"query_id":"q1","workflow_id":"wf","status":"pending"}"#; + const STREAM_PATH: &str = "/v1/advisor/query/stream"; + + const PHASE_FETCHING: &str = r#"{"type":"phase","data":{"phase":"fetching"}}"#; + const PHASE_INTERPRETING: &str = r#"{"type":"phase","data":{"phase":"interpreting"}}"#; + const PHASE_SUMMARIZING: &str = r#"{"type":"phase","data":{"phase":"summarizing"}}"#; + const HEARTBEAT: &str = r#"{"type":"heartbeat","data":{"timestamp":1}}"#; + const FINAL_NO_ADRS: &str = r#"{"type":"final","data":{"summary":"Use the App Router.","interpreter":{"summary":"i","related_adrs":[]}}}"#; + const ONE_ADR: &str = r#"{"id":"a1","name":"n","title":"Use the App Router","policy":"p","instructions":"i","scope":"frontend","relevance_reason":"r","confidence":0.92}"#; + + fn final_with_adr() -> String { + format!( + r#"{{"type":"final","data":{{"summary":"Use the App Router.","interpreter":{{"summary":"i","related_adrs":[{ONE_ADR}]}}}}}}"# + ) + } + + /// Build an SSE body from a list of JSON frame payloads. + fn sse_body(frames: &[&str]) -> String { + frames.iter().map(|f| format!("data: {f}\n\n")).collect() + } fn test_creds() -> StoredCredentials { StoredCredentials { @@ -262,20 +409,13 @@ mod tests { } } - fn succeeded_body(adrs_json: &str) -> String { - format!( - r#"{{"query_id":"q1","status":"succeeded","result":{{"summary":"Use the App Router.","interpreter":{{"summary":"i","related_adrs":[{adrs_json}]}}}},"error":null}}"# - ) - } - - const ONE_ADR: &str = r#"{"id":"a1","name":"n","title":"Use the App Router","policy":"p","instructions":"i","scope":"frontend","relevance_reason":"r","confidence":0.92}"#; - fn args(api_url: &str, org: Option<&str>) -> AdvisorArgs { AdvisorArgs { query: "why app router?".to_string(), api_url: Some(api_url.to_string()), org: org.map(|s| s.to_string()), repo: None, + json: false, } } @@ -336,6 +476,229 @@ mod tests { print_answer(&without); } + #[test] + fn test_render_phase_silent_and_fallback_arms() { + // `done` and the empty phase are silent (the `final` frame carries the + // answer, so there is nothing to print for them); an unrecognized phase + // falls through to the generic one-line render. None of these panic — + // exercising each arm locks the rendering contract in. + render_phase("done"); + render_phase(""); + render_phase("compiling"); + } + + // --- SSE frame decoding (chunk reassembly) --- + + #[test] + fn test_decoder_single_frame_one_push() { + let mut d = SseDecoder::default(); + let out = d.push(b"data: {\"type\":\"heartbeat\"}\n\n"); + assert_eq!(out, vec!["{\"type\":\"heartbeat\"}".to_string()]); + } + + #[test] + fn test_decoder_multiple_frames_one_push() { + let mut d = SseDecoder::default(); + let body = sse_body(&[PHASE_FETCHING, FINAL_NO_ADRS]); + let out = d.push(body.as_bytes()); + assert_eq!(out.len(), 2); + assert_eq!(out[0], PHASE_FETCHING); + assert_eq!(out[1], FINAL_NO_ADRS); + } + + #[test] + fn test_decoder_reassembles_frame_split_across_chunks() { + let mut d = SseDecoder::default(); + let frame = format!("data: {FINAL_NO_ADRS}\n\n"); + let bytes = frame.as_bytes(); + let mid = bytes.len() / 2; + // First half: no complete frame yet. + assert!(d.push(&bytes[..mid]).is_empty()); + // Second half completes the frame. + let out = d.push(&bytes[mid..]); + assert_eq!(out, vec![FINAL_NO_ADRS.to_string()]); + } + + #[test] + fn test_decoder_reassembles_frame_split_byte_by_byte() { + let mut d = SseDecoder::default(); + let frame = format!("data: {PHASE_FETCHING}\n\n"); + let mut collected = Vec::new(); + for b in frame.as_bytes() { + collected.extend(d.push(&[*b])); + } + assert_eq!(collected, vec![PHASE_FETCHING.to_string()]); + } + + #[test] + fn test_decoder_handles_crlf_separators() { + let mut d = SseDecoder::default(); + let out = d.push(b"data: {\"type\":\"heartbeat\"}\r\n\r\n"); + assert_eq!(out, vec!["{\"type\":\"heartbeat\"}".to_string()]); + } + + #[test] + fn test_decoder_ignores_comment_only_event() { + let mut d = SseDecoder::default(); + // A comment-only keep-alive (no `data:` line) yields nothing. + let out = d.push(b": keep-alive\n\n"); + assert!(out.is_empty()); + } + + #[test] + fn test_decoder_flush_returns_unterminated_trailing_event() { + let mut d = SseDecoder::default(); + // No trailing blank line — push yields nothing, flush recovers it. + assert!(d.push(b"data: {\"type\":\"final\"}").is_empty()); + assert_eq!(d.flush(), Some("{\"type\":\"final\"}".to_string())); + // Buffer is drained after flush. + assert_eq!(d.flush(), None); + } + + #[test] + fn test_decoder_strips_optional_space_after_prefix() { + let mut d = SseDecoder::default(); + // SSE allows `data:x` (no space) and `data: x` (one space) — both strip + // to `x`. + assert_eq!(d.push(b"data:no-space\n\n"), vec!["no-space".to_string()]); + assert_eq!( + d.push(b"data: one-space\n\n"), + vec!["one-space".to_string()] + ); + } + + // --- frame parsing --- + + #[test] + fn test_parse_frame_phase() { + assert_eq!( + parse_advisor_frame(PHASE_FETCHING), + Some(AdvisorStreamEvent::Phase("fetching".to_string())) + ); + } + + #[test] + fn test_parse_frame_final_carries_output() { + // `final` parses to a Final event carrying the verbatim `data` object. + // Asserting equality (rather than matching with a fallback arm) keeps the + // test free of an untaken branch; the asserted-equal value is then what + // deserializes into an AdvisorOutput. + let event = parse_advisor_frame(FINAL_NO_ADRS).unwrap(); + let expected: serde_json::Value = serde_json::from_str( + r#"{"summary":"Use the App Router.","interpreter":{"summary":"i","related_adrs":[]}}"#, + ) + .unwrap(); + assert_eq!(event, AdvisorStreamEvent::Final(expected.clone())); + let out: AdvisorOutput = serde_json::from_value(expected).unwrap(); + assert_eq!(out.summary, "Use the App Router."); + assert!(out.interpreter.related_adrs.is_empty()); + } + + #[test] + fn test_parse_frame_error() { + let frame = r#"{"type":"error","error":"stream ended"}"#; + assert_eq!( + parse_advisor_frame(frame), + Some(AdvisorStreamEvent::Error("stream ended".to_string())) + ); + } + + #[test] + fn test_parse_frame_heartbeat() { + assert_eq!( + parse_advisor_frame(HEARTBEAT), + Some(AdvisorStreamEvent::Heartbeat) + ); + } + + #[test] + fn test_parse_frame_unknown_type_is_other() { + let frame = r#"{"type":"surprise","data":{}}"#; + assert_eq!( + parse_advisor_frame(frame), + Some(AdvisorStreamEvent::Other("surprise".to_string())) + ); + } + + #[test] + fn test_parse_frame_non_json_is_none() { + assert_eq!(parse_advisor_frame("not json"), None); + assert_eq!(parse_advisor_frame(""), None); + // A JSON value that is not an object, or lacks a string `type`. + assert_eq!(parse_advisor_frame("[1,2,3]"), None); + assert_eq!(parse_advisor_frame(r#"{"no":"type"}"#), None); + } + + // --- per-frame dispatch (handle_payload) --- + + #[test] + fn test_handle_payload_phase_is_non_terminal() { + assert_eq!(handle_payload(PHASE_FETCHING).unwrap(), None); + } + + #[test] + fn test_handle_payload_heartbeat_is_ignored() { + assert_eq!(handle_payload(HEARTBEAT).unwrap(), None); + } + + #[test] + fn test_handle_payload_unknown_and_garbage_are_ignored() { + assert_eq!(handle_payload(r#"{"type":"surprise"}"#).unwrap(), None); + assert_eq!(handle_payload("not json").unwrap(), None); + } + + #[test] + fn test_handle_payload_final_is_terminal() { + // A `final` payload is the one terminal success outcome; assert equality + // rather than matching so there is no untaken arm left uncovered. + let outcome = handle_payload(FINAL_NO_ADRS).unwrap(); + let expected: serde_json::Value = serde_json::from_str( + r#"{"summary":"Use the App Router.","interpreter":{"summary":"i","related_adrs":[]}}"#, + ) + .unwrap(); + assert_eq!(outcome, Some(StreamOutcome::Final(expected.clone()))); + let out: AdvisorOutput = serde_json::from_value(expected).unwrap(); + assert_eq!(out.summary, "Use the App Router."); + } + + #[test] + fn test_handle_payload_error_frame_is_error() { + let frame = r#"{"type":"error","error":"boom"}"#; + let err = handle_payload(frame).unwrap_err(); + assert!(matches!(err, ActualError::ApiError(ref m) if m.contains("boom"))); + } + + // --- print_final --- + + #[test] + fn test_print_final_json_pretty_prints() { + let data = serde_json::json!({ + "summary": "S", + "interpreter": {"summary": "i", "related_adrs": []} + }); + // --json echoes the object verbatim; valid input never errors. + print_final(&data, true).unwrap(); + } + + #[test] + fn test_print_final_human_renders_valid_output() { + let data = serde_json::json!({ + "summary": "S", + "interpreter": {"summary": "i", "related_adrs": []} + }); + print_final(&data, false).unwrap(); + } + + #[test] + fn test_print_final_human_rejects_unparseable_output() { + // Missing `summary` → not a valid AdvisorOutput → error on the human path. + let data = serde_json::json!({"interpreter": {"summary": "i"}}); + let err = print_final(&data, false).unwrap_err(); + assert!(matches!(err, ActualError::ApiError(ref m) if m.contains("unparseable"))); + } + + // --- end-to-end via a mocked stream --- + #[test] fn test_exec_not_logged_in() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); @@ -352,18 +715,12 @@ mod tests { let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); let _g2 = EnvGuard::set("ACTUAL_CONFIG_DIR", tmp.path().to_str().unwrap()); - let err = run( - &args("http://127.0.0.1:1", None), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap_err(); + let err = run(&args("http://127.0.0.1:1", None)).await.unwrap_err(); assert!(matches!(err, ActualError::NotLoggedIn)); } #[tokio::test] - async fn test_run_success_renders_related_adrs() { + async fn test_run_streams_phases_then_final() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); @@ -371,35 +728,26 @@ mod tests { store::save(&test_creds()).unwrap(); let mut server = mockito::Server::new_async().await; - let s = server - .mock("POST", "/v1/advisor/query") + let body = sse_body(&[ + PHASE_FETCHING, + PHASE_INTERPRETING, + PHASE_SUMMARIZING, + &final_with_adr(), + ]); + let m = server + .mock("POST", STREAM_PATH) .with_status(200) - .with_header("content-type", "application/json") - .with_body(START_BODY) - .create_async() - .await; - let p = server - .mock("GET", POLL_PATH) - .with_status(200) - .with_header("content-type", "application/json") - .with_body(succeeded_body(ONE_ADR)) + .with_header("content-type", "text/event-stream") + .with_body(body) .create_async() .await; - // org omitted → uses the signed-in org from creds. - run( - &args(&server.url(), None), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap(); - s.assert_async().await; - p.assert_async().await; + run(&args(&server.url(), None)).await.unwrap(); + m.assert_async().await; } #[tokio::test] - async fn test_run_success_no_related_adrs() { + async fn test_run_ignores_interleaved_heartbeats() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); @@ -407,62 +755,20 @@ mod tests { store::save(&test_creds()).unwrap(); let mut server = mockito::Server::new_async().await; - let _s = server - .mock("POST", "/v1/advisor/query") - .with_body(START_BODY) - .with_header("content-type", "application/json") - .create_async() - .await; - let _p = server - .mock("GET", POLL_PATH) - .with_body(succeeded_body("")) - .with_header("content-type", "application/json") + let body = sse_body(&[HEARTBEAT, PHASE_FETCHING, HEARTBEAT, FINAL_NO_ADRS]); + let _m = server + .mock("POST", STREAM_PATH) + .with_status(200) + .with_header("content-type", "text/event-stream") + .with_body(body) .create_async() .await; - run( - &args(&server.url(), Some("00000000-0000-0000-0000-000000000000")), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap(); - } - - #[tokio::test] - async fn test_run_failed_query() { - let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); - let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); - let tmp = tempdir().unwrap(); - let _g2 = EnvGuard::set("ACTUAL_CONFIG_DIR", tmp.path().to_str().unwrap()); - store::save(&test_creds()).unwrap(); - let mut server = mockito::Server::new_async().await; - let _s = server - .mock("POST", "/v1/advisor/query") - .with_body(START_BODY) - .with_header("content-type", "application/json") - .create_async() - .await; - let _p = server - .mock("GET", POLL_PATH) - .with_body( - r#"{"query_id":"q1","status":"failed","result":null,"error":"stream ended"}"#, - ) - .with_header("content-type", "application/json") - .create_async() - .await; - let err = run( - &args(&server.url(), None), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap_err(); - assert!(matches!(err, ActualError::ApiError(ref m) if m.contains("stream ended"))); + run(&args(&server.url(), None)).await.unwrap(); } #[tokio::test] - async fn test_run_succeeded_without_result_is_failure() { + async fn test_run_json_output() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); @@ -470,30 +776,22 @@ mod tests { store::save(&test_creds()).unwrap(); let mut server = mockito::Server::new_async().await; - let _s = server - .mock("POST", "/v1/advisor/query") - .with_body(START_BODY) - .with_header("content-type", "application/json") - .create_async() - .await; - let _p = server - .mock("GET", POLL_PATH) - .with_body(r#"{"query_id":"q1","status":"succeeded","result":null,"error":null}"#) - .with_header("content-type", "application/json") + let body = sse_body(&[PHASE_FETCHING, FINAL_NO_ADRS]); + let _m = server + .mock("POST", STREAM_PATH) + .with_status(200) + .with_header("content-type", "text/event-stream") + .with_body(body) .create_async() .await; - let err = run( - &args(&server.url(), None), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap_err(); - assert!(matches!(err, ActualError::ApiError(ref m) if m.contains("no result"))); + + let mut a = args(&server.url(), None); + a.json = true; + run(&a).await.unwrap(); } #[tokio::test] - async fn test_run_running_then_succeeded() { + async fn test_run_error_frame_is_non_zero_exit() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); @@ -501,76 +799,26 @@ mod tests { store::save(&test_creds()).unwrap(); let mut server = mockito::Server::new_async().await; - let _s = server - .mock("POST", "/v1/advisor/query") - .with_body(START_BODY) - .with_header("content-type", "application/json") - .create_async() - .await; - // First poll: running (Retry-After: 0 → immediate). Second: succeeded. - let _running = server - .mock("GET", POLL_PATH) + let body = sse_body(&[ + PHASE_FETCHING, + r#"{"type":"error","error":"model unavailable"}"#, + ]); + let _m = server + .mock("POST", STREAM_PATH) .with_status(200) - .with_header("content-type", "application/json") - .with_header("retry-after", "0") - .with_body(r#"{"query_id":"q1","status":"running","result":null,"error":null}"#) - .expect(1) - .create_async() - .await; - let _done = server - .mock("GET", POLL_PATH) - .with_body(succeeded_body(ONE_ADR)) - .with_header("content-type", "application/json") + .with_header("content-type", "text/event-stream") + .with_body(body) .create_async() .await; - // org provided via --org (exercises the args.org branch). - run( - &args(&server.url(), Some("22222222-2222-2222-2222-222222222222")), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap(); - } - #[tokio::test] - async fn test_run_not_modified_then_succeeded() { - let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); - let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); - let tmp = tempdir().unwrap(); - let _g2 = EnvGuard::set("ACTUAL_CONFIG_DIR", tmp.path().to_str().unwrap()); - store::save(&test_creds()).unwrap(); - - let mut server = mockito::Server::new_async().await; - let _s = server - .mock("POST", "/v1/advisor/query") - .with_body(START_BODY) - .with_header("content-type", "application/json") - .create_async() - .await; - let _nm = server - .mock("GET", POLL_PATH) - .with_status(304) - .expect(1) - .create_async() - .await; - let _done = server - .mock("GET", POLL_PATH) - .with_body(succeeded_body("")) - .with_header("content-type", "application/json") - .create_async() - .await; - run( - &args(&server.url(), None), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap(); + let err = run(&args(&server.url(), None)).await.unwrap_err(); + assert!(matches!(err, ActualError::ApiError(ref m) if m.contains("model unavailable"))); + // ApiError is a non-zero exit. + assert_ne!(err.exit_code(), 0); } #[tokio::test] - async fn test_poll_times_out_at_deadline() { + async fn test_run_graceful_close_without_final_is_completion() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); @@ -578,34 +826,22 @@ mod tests { store::save(&test_creds()).unwrap(); let mut server = mockito::Server::new_async().await; - let _s = server - .mock("POST", "/v1/advisor/query") - .with_body(START_BODY) - .with_header("content-type", "application/json") - .create_async() - .await; - // Always running → the loop keeps polling until the wall-clock deadline. - let _p = server - .mock("GET", POLL_PATH) + // Phases only, then the server closes — no `final`, no `error`. + let body = sse_body(&[PHASE_FETCHING, PHASE_INTERPRETING]); + let _m = server + .mock("POST", STREAM_PATH) .with_status(200) - .with_header("content-type", "application/json") - .with_header("retry-after", "0") - .with_body(r#"{"query_id":"q1","status":"running","result":null,"error":null}"#) + .with_header("content-type", "text/event-stream") + .with_body(body) .create_async() .await; - // Tiny deadline + zero interval → polls a few times, then gives up. - let err = run( - &args(&server.url(), None), - Duration::from_millis(10), - Duration::ZERO, - ) - .await - .unwrap_err(); - assert!(matches!(err, ActualError::ApiError(ref m) if m.contains("did not reach"))); + + // Graceful close with no final frame is completion, not an error. + run(&args(&server.url(), None)).await.unwrap(); } #[tokio::test] - async fn test_run_sends_versioned_job_envelope() { + async fn test_run_final_recovered_via_flush_without_trailing_blank_line() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); @@ -613,38 +849,24 @@ mod tests { store::save(&test_creds()).unwrap(); let mut server = mockito::Server::new_async().await; - // The server validates the typed/versioned envelope: type + version - // literals and the query nested under `data`. - let s = server - .mock("POST", "/v1/advisor/query") - .match_body(mockito::Matcher::PartialJsonString( - r#"{"type":"advisor_query","version":1,"data":{"query":"why app router?"}}"# - .to_string(), - )) + // A blank-line-terminated phase frame (consumed via `push`) followed by a + // `final` frame the server closes WITHOUT the terminating blank line, so + // the answer is only recovered by the end-of-stream `flush`. + let body = format!("data: {PHASE_FETCHING}\n\ndata: {FINAL_NO_ADRS}"); + let _m = server + .mock("POST", STREAM_PATH) .with_status(200) - .with_header("content-type", "application/json") - .with_body(START_BODY) - .create_async() - .await; - let _p = server - .mock("GET", POLL_PATH) - .with_body(succeeded_body("")) - .with_header("content-type", "application/json") + .with_header("content-type", "text/event-stream") + .with_body(body) .create_async() .await; - run( - &args(&server.url(), None), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap(); - s.assert_async().await; + // The flush-recovered `final` frame is a successful terminal outcome. + run(&args(&server.url(), None)).await.unwrap(); } #[tokio::test] - async fn test_run_retries_on_transient_500() { + async fn test_run_flush_recovers_non_terminal_frame_then_closes() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); @@ -652,117 +874,43 @@ mod tests { store::save(&test_creds()).unwrap(); let mut server = mockito::Server::new_async().await; - let _s = server - .mock("POST", "/v1/advisor/query") - .with_body(START_BODY) - .with_header("content-type", "application/json") - .create_async() - .await; - // First poll: transient infra 500 → retried, not fatal. Second: succeeded. - let infra = server - .mock("GET", POLL_PATH) - .with_status(500) - .with_body(r#"{"error":"row load failed"}"#) - .expect(1) - .create_async() - .await; - let _done = server - .mock("GET", POLL_PATH) - .with_body(succeeded_body(ONE_ADR)) - .with_header("content-type", "application/json") + // The stream closes right after a phase frame, with no terminating blank + // line: the phase is recovered by `flush` but is non-terminal, so the run + // ends as a graceful close (no `final`, no `error`) rather than returning + // a flushed answer. + let body = format!("data: {PHASE_FETCHING}"); + let _m = server + .mock("POST", STREAM_PATH) + .with_status(200) + .with_header("content-type", "text/event-stream") + .with_body(body) .create_async() .await; - run( - &args(&server.url(), None), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap(); - infra.assert_async().await; - } - - #[test] - fn test_next_delay_clamps_retry_after() { - // A large (or misbehaving) server Retry-After is clamped to the ceiling. - assert_eq!( - next_delay(Some(600), Duration::from_secs(2)), - MAX_RETRY_AFTER - ); - assert_eq!( - next_delay(Some(15), Duration::from_secs(2)), - Duration::from_secs(15) - ); - // Values under the ceiling pass through; None falls back to the default. - assert_eq!( - next_delay(Some(3), Duration::from_secs(2)), - Duration::from_secs(3) - ); - assert_eq!( - next_delay(None, Duration::from_secs(2)), - Duration::from_secs(2) - ); - } - - // --- transparent refresh-on-expiry --- - - #[tokio::test] - async fn test_ensure_fresh_no_refresh_token_returns_unchanged() { - let mut c = test_creds(); - c.refresh_token = String::new(); - let out = ensure_fresh(c.clone()).await.unwrap(); - assert_eq!(out.access_token, c.access_token); + run(&args(&server.url(), None)).await.unwrap(); } #[tokio::test] - async fn test_ensure_fresh_not_expired_returns_unchanged() { - let mut c = test_creds(); - c.expires_at = Some(Utc::now() + ChronoDuration::hours(1)); - let out = ensure_fresh(c).await.unwrap(); - assert_eq!(out.access_token, "tok"); - } - - #[tokio::test] - async fn test_ensure_fresh_expired_refreshes_and_persists() { + async fn test_run_non_2xx_is_non_zero_exit() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); let tmp = tempdir().unwrap(); let _g2 = EnvGuard::set("ACTUAL_CONFIG_DIR", tmp.path().to_str().unwrap()); + store::save(&test_creds()).unwrap(); let mut server = mockito::Server::new_async().await; let _m = server - .mock("POST", "/api/oauth/token") - .with_status(200) + .mock("POST", STREAM_PATH) + .with_status(400) .with_header("content-type", "application/json") - .with_body( - r#"{"access_token":"new-at","token_type":"Bearer","expires_in":3600,"refresh_token":"new-rt"}"#, - ) + .with_body(r#"{"error":{"code":"BAD_REQUEST","message":"bad","details":null}}"#) .create_async() .await; - let mut c = test_creds(); - c.expires_at = Some(Utc::now() - ChronoDuration::seconds(1)); // expired - c.auth_url = Some(server.url()); - - let out = ensure_fresh(c).await.unwrap(); - assert_eq!(out.access_token, "new-at"); - // Rotated creds were re-persisted. - assert_eq!(store::load().unwrap().unwrap().access_token, "new-at"); + let err = run(&args(&server.url(), None)).await.unwrap_err(); + assert_ne!(err.exit_code(), 0); } - #[tokio::test] - async fn test_ensure_fresh_refresh_failure_is_not_logged_in() { - // Expired + unreachable auth server → refresh errors → NotLoggedIn. - let mut c = test_creds(); - c.expires_at = Some(Utc::now() - ChronoDuration::seconds(1)); - c.auth_url = Some("http://127.0.0.1:1".to_string()); - let err = ensure_fresh(c).await.unwrap_err(); - assert!(matches!(err, ActualError::NotLoggedIn)); - } - - // --- explicit repo scoping --- - #[tokio::test] async fn test_run_with_explicit_repo_scopes_request() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); @@ -773,32 +921,22 @@ mod tests { let mut server = mockito::Server::new_async().await; let s = server - .mock("POST", "/v1/advisor/query") + .mock("POST", STREAM_PATH) .match_body(mockito::Matcher::PartialJsonString( r#"{"repo_unique_id":"33333333-3333-3333-3333-333333333333"}"#.to_string(), )) .with_status(200) - .with_header("content-type", "application/json") - .with_body(START_BODY) - .create_async() - .await; - let _p = server - .mock("GET", POLL_PATH) - .with_body(succeeded_body("")) - .with_header("content-type", "application/json") + .with_header("content-type", "text/event-stream") + .with_body(sse_body(&[FINAL_NO_ADRS])) .create_async() .await; let mut a = args(&server.url(), None); a.repo = Some("33333333-3333-3333-3333-333333333333".to_string()); - run(&a, Duration::from_secs(60), Duration::ZERO) - .await - .unwrap(); + run(&a).await.unwrap(); s.assert_async().await; } - // --- cross-org 403 handling --- - #[tokio::test] async fn test_run_org_mismatch_403_surfaces_actionable_error() { let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); @@ -810,7 +948,7 @@ mod tests { let mut server = mockito::Server::new_async().await; // api-service rejects the cross-org token with a fail-closed 403. let _s = server - .mock("POST", "/v1/advisor/query") + .mock("POST", STREAM_PATH) .with_status(403) .with_header("content-type", "application/json") .with_body(r#"{"error":{"code":"FORBIDDEN","message":"cross-org","details":null}}"#) @@ -819,13 +957,7 @@ mod tests { // Explicit --org that differs from the session org (test_creds is 1111…). let target = "99999999-9999-9999-9999-999999999999"; - let err = run( - &args(&server.url(), Some(target)), - Duration::from_secs(60), - Duration::ZERO, - ) - .await - .unwrap_err(); + let err = run(&args(&server.url(), Some(target))).await.unwrap_err(); match err { ActualError::OrgMismatch { message, hint } => { @@ -889,4 +1021,60 @@ mod tests { "remediation should not be in the message: {message}" ); } + + // --- transparent refresh-on-expiry --- + + #[tokio::test] + async fn test_ensure_fresh_no_refresh_token_returns_unchanged() { + let mut c = test_creds(); + c.refresh_token = String::new(); + let out = ensure_fresh(c.clone()).await.unwrap(); + assert_eq!(out.access_token, c.access_token); + } + + #[tokio::test] + async fn test_ensure_fresh_not_expired_returns_unchanged() { + let mut c = test_creds(); + c.expires_at = Some(Utc::now() + ChronoDuration::hours(1)); + let out = ensure_fresh(c).await.unwrap(); + assert_eq!(out.access_token, "tok"); + } + + #[tokio::test] + async fn test_ensure_fresh_expired_refreshes_and_persists() { + let _lock = ENV_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); + let _g1 = EnvGuard::remove("ACTUAL_CONFIG"); + let tmp = tempdir().unwrap(); + let _g2 = EnvGuard::set("ACTUAL_CONFIG_DIR", tmp.path().to_str().unwrap()); + + let mut server = mockito::Server::new_async().await; + let _m = server + .mock("POST", "/api/oauth/token") + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + r#"{"access_token":"new-at","token_type":"Bearer","expires_in":3600,"refresh_token":"new-rt"}"#, + ) + .create_async() + .await; + + let mut c = test_creds(); + c.expires_at = Some(Utc::now() - ChronoDuration::seconds(1)); // expired + c.auth_url = Some(server.url()); + + let out = ensure_fresh(c).await.unwrap(); + assert_eq!(out.access_token, "new-at"); + // Rotated creds were re-persisted. + assert_eq!(store::load().unwrap().unwrap().access_token, "new-at"); + } + + #[tokio::test] + async fn test_ensure_fresh_refresh_failure_is_not_logged_in() { + // Expired + unreachable auth server → refresh errors → NotLoggedIn. + let mut c = test_creds(); + c.expires_at = Some(Utc::now() - ChronoDuration::seconds(1)); + c.auth_url = Some("http://127.0.0.1:1".to_string()); + let err = ensure_fresh(c).await.unwrap_err(); + assert!(matches!(err, ActualError::NotLoggedIn)); + } }