diff --git a/crates/crw-mcp-proto/src/lib.rs b/crates/crw-mcp-proto/src/lib.rs index 58ac8f3a..981ab2c8 100644 --- a/crates/crw-mcp-proto/src/lib.rs +++ b/crates/crw-mcp-proto/src/lib.rs @@ -88,6 +88,61 @@ impl JsonRpcResponse { // --- Tool definitions --- +fn extract_accepted_output_schema() -> Value { + json!({ + "type": "object", + "additionalProperties": false, + "properties": { + "success": { "type": "boolean" }, + "id": { "type": "string" }, + "status": { "type": "string", "enum": ["processing"] }, + "urls": { "type": "integer", "minimum": 0 } + }, + "required": ["success", "id", "status", "urls"] + }) +} + +fn extract_status_output_schema() -> Value { + json!({ + "type": "object", + "additionalProperties": false, + "properties": { + "success": { "type": "boolean" }, + "id": { "type": "string" }, + "status": { + "type": "string", + "enum": ["processing", "cancelling", "completed", "failed", "cancelled"] + }, + "results": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "properties": { + "url": { "type": "string" }, + "status": { + "type": "string", + "enum": ["processing", "completed", "failed", "cancelled"] + }, + "data": { "type": "object", "additionalProperties": true }, + "error": { "type": "string" }, + "llmUsage": { "type": "object" }, + "basis": { "type": "array", "items": { "type": "object" } }, + "basisWarnings": { "type": "array", "items": { "type": "object" } }, + "llmInputHash": { "type": "string" } + }, + "required": ["url", "status"] + } + }, + "error": { "type": "string" }, + "expiresAt": { "type": "string", "format": "date-time" }, + "creditsUsed": { "type": "integer" }, + "tokensUsed": { "type": "integer" } + }, + "required": ["success", "id", "status", "results", "expiresAt", "creditsUsed", "tokensUsed"] + }) +} + pub fn tool_definitions(proxy_mode: bool) -> Value { let mut tools = vec![ json!({ @@ -298,7 +353,8 @@ pub fn tool_definitions(proxy_mode: bool) -> Value { "llmModel": { "type": "string" } }, "required": ["urls"] - } + }, + "outputSchema": extract_accepted_output_schema() }), json!({ "name": "crw_check_extract_status", @@ -316,7 +372,27 @@ pub fn tool_definitions(proxy_mode: bool) -> Value { "id": { "type": "string", "description": "Extract job id from crw_extract" } }, "required": ["id"] - } + }, + "outputSchema": extract_status_output_schema() + }), + json!({ + "name": "crw_cancel_extract", + "title": "Cancel extract job", + "description": "Request cancellation of an extract job. Returns the canonical status; cancelling remains non-terminal until the claimed URL settles.", + "annotations": { + "readOnlyHint": false, + "destructiveHint": true, + "idempotentHint": true, + "openWorldHint": true + }, + "inputSchema": { + "type": "object", + "properties": { + "id": { "type": "string", "description": "Extract job id from crw_extract" } + }, + "required": ["id"] + }, + "outputSchema": extract_status_output_schema() }), ]; @@ -821,12 +897,12 @@ mod tests { /// /// Baseline before the Phase 1 trim was 8233 bytes (~2744 est-tok). After the /// Phase 1 trim + Phase 3 annotations/titles + the two native extract tools the - /// full 8-tool list is ~8017 bytes (~2673 est-tok). The ceiling is floor + ~3% - /// headroom so the gate catches real bloat without churning on minor edits. - // Raised from 2300 when the two native extract tools (crw_extract + - // crw_check_extract_status) were added — proportional to the existing - // per-tool footprint, not description bloat. - const TOOLS_LIST_TOKEN_CEILING: usize = 2750; + /// full 8-tool list was ~8017 bytes (~2673 est-tok). The canonical lifecycle + /// adds one cancel tool plus required output schemas for start/status/cancel; + /// after closing lifecycle statuses and typing every per-URL result field, + /// the 9-tool list is ~10705 bytes (~3569 est-tok). The ceiling keeps ~2% + /// headroom so further growth still fails. + const TOOLS_LIST_TOKEN_CEILING: usize = 3650; #[test] fn tools_list_token_budget() { @@ -1306,20 +1382,14 @@ mod tests { } /// A1 — every tool advertises annotations + a title; crw_crawl and crw_extract - /// are the non-read-only / non-idempotent tools; crw_parse_file is the only closed-world. + /// are non-idempotent, while cancel is destructive but idempotent. #[test] fn a1_tools_advertise_annotations_and_title() { let defs = tool_definitions(false); for t in defs["tools"].as_array().unwrap() { assert!(t["annotations"].is_object(), "{} annotations", t["name"]); assert!(t["title"].is_string(), "{} title", t["name"]); - // destructiveHint is explicitly false everywhere (the JSON default is true). - assert_eq!( - t["annotations"]["destructiveHint"], - json!(false), - "{}", - t["name"] - ); + assert!(t["annotations"]["destructiveHint"].is_boolean()); } let crawl = tool_by_name(&defs, "crw_crawl"); assert_eq!(crawl["annotations"]["readOnlyHint"], json!(false)); @@ -1328,6 +1398,10 @@ mod tests { let extract = tool_by_name(&defs, "crw_extract"); assert_eq!(extract["annotations"]["readOnlyHint"], json!(false)); assert_eq!(extract["annotations"]["idempotentHint"], json!(false)); + let cancel = tool_by_name(&defs, "crw_cancel_extract"); + assert_eq!(cancel["annotations"]["readOnlyHint"], json!(false)); + assert_eq!(cancel["annotations"]["destructiveHint"], json!(true)); + assert_eq!(cancel["annotations"]["idempotentHint"], json!(true)); let scrape = tool_by_name(&defs, "crw_scrape"); assert_eq!(scrape["annotations"]["readOnlyHint"], json!(true)); assert_eq!(scrape["annotations"]["openWorldHint"], json!(true)); @@ -1335,7 +1409,7 @@ mod tests { assert_eq!(parse["annotations"]["openWorldHint"], json!(false)); } - /// A2 — is_known_tool recognizes all 8 tool names, rejects others. + /// A2 — is_known_tool recognizes all 9 tool names, rejects others. #[test] fn a2_is_known_tool() { for name in [ @@ -1347,6 +1421,7 @@ mod tests { "crw_parse_file", "crw_extract", "crw_check_extract_status", + "crw_cancel_extract", ] { assert!(is_known_tool(name), "{name} should be known"); } @@ -1378,10 +1453,10 @@ mod tests { } let with = list(true); assert!(with.contains(&"crw_search".to_string())); - assert_eq!(with.len(), 8); + assert_eq!(with.len(), 9); let without = list(false); assert!(!without.contains(&"crw_search".to_string())); - assert_eq!(without.len(), 7); + assert_eq!(without.len(), 8); } /// A4 — initialize advertises server usage `instructions` (the model-facing diff --git a/crates/crw-mcp/README.md b/crates/crw-mcp/README.md index 0233d3c9..71fbc643 100644 --- a/crates/crw-mcp/README.md +++ b/crates/crw-mcp/README.md @@ -26,6 +26,7 @@ MCP (Model Context Protocol) server for the [CRW](https://github.com/us/crw) web | `crw_map` | Discover all URLs on a website | | `crw_extract` | Extract structured JSON from URLs via a prompt and/or JSON schema (async job, returns job ID) | | `crw_check_extract_status` | Poll extract job status and retrieve results | +| `crw_cancel_extract` | Idempotently cancel an extract job and retrieve canonical status | | `crw_search` | Search the web (needs a configured search backend; always available in proxy mode, available in embedded mode only when a search backend is configured) | | `crw_parse_file` | Parse a local PDF (base64) to markdown | diff --git a/crates/crw-mcp/src/main.rs b/crates/crw-mcp/src/main.rs index 8d0d791e..2a5bd24d 100644 --- a/crates/crw-mcp/src/main.rs +++ b/crates/crw-mcp/src/main.rs @@ -17,6 +17,7 @@ //! - `crw_search` — web search (embedded: only when a search backend is configured; proxy: forwards to the remote API) //! - `crw_extract` — start an async multi-URL structured extraction job //! - `crw_check_extract_status` — poll extract job status +//! - `crw_cancel_extract` — idempotently cancel an extract job //! - `crw_parse_file` — parse a local PDF (base64) to markdown //! //! # Usage @@ -261,6 +262,8 @@ async fn proxy_call_tool( parse_response(resp).await } "crw_extract" => { + let mut headers = headers; + headers.insert("prefer", "respond-async".parse().unwrap()); let resp = client .post(format!("{base_url}/v1/extract")) .headers(headers) @@ -285,6 +288,20 @@ async fn proxy_call_tool( .map_err(|e| format!("HTTP request failed: {e}"))?; parse_response(resp).await } + "crw_cancel_extract" => { + let id = args + .get("id") + .and_then(|v| v.as_str()) + .ok_or("missing required parameter: id")?; + let resp = client + .delete(format!("{base_url}/v1/extract/{id}")) + .headers(headers) + .timeout(TIMEOUT_CRAWL_STATUS) + .send() + .await + .map_err(|e| format!("HTTP request failed: {e}"))?; + parse_response(resp).await + } "crw_parse_file" => { use base64::Engine; let b64 = args diff --git a/crates/crw-server/openapi/openapi-3.0.json b/crates/crw-server/openapi/openapi-3.0.json index 55d30575..25ae5aa6 100644 --- a/crates/crw-server/openapi/openapi-3.0.json +++ b/crates/crw-server/openapi/openapi-3.0.json @@ -406,8 +406,22 @@ "extract" ], "summary": "Start an async multi-URL structured extraction job", - "description": "Native extraction. Returns a per-URL array via GET /v1/extract/{id}. At least one of `prompt` or `schema` is required.", + "description": "Native extraction. Returns a fixed-cardinality per-URL array via GET /v1/extract/{id}. At least one of `prompt` or `schema` is required. SDK async starters send `Prefer: respond-async` for managed/self-hosted parity.", "operationId": "extractStart", + "parameters": [ + { + "name": "Prefer", + "in": "header", + "required": false, + "schema": { + "type": "string", + "enum": [ + "respond-async" + ] + }, + "description": "Request the canonical async lifecycle across managed and self-hosted deployments." + } + ], "requestBody": { "required": true, "content": { @@ -464,6 +478,46 @@ } } }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "404": { + "$ref": "#/components/responses/NotFound" + } + } + }, + "delete": { + "tags": [ + "extract" + ], + "summary": "Cancel an extract job", + "description": "Idempotent. Processing becomes cancelling while a claimed URL settles, then cancelled. Completed, failed, and cancelled states are immutable.", + "operationId": "extractCancel", + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "description": "Extract job id returned by POST /v1/extract" + } + ], + "responses": { + "200": { + "description": "Canonical job status + ordered per-URL results", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ExtractStatus" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, "404": { "$ref": "#/components/responses/NotFound" } @@ -1386,14 +1440,16 @@ "status": { "type": "string", "enum": [ + "processing", "completed", - "failed" + "failed", + "cancelled" ] }, "data": { "type": "object", "additionalProperties": true, - "description": "Extracted JSON object (present when status is completed)." + "description": "Extracted JSON object (present when status is completed). Untouched cancelled slots omit data, error, usage, and basis." }, "error": { "type": "string" @@ -1427,6 +1483,7 @@ "required": [ "id", "status", + "results", "expiresAt", "creditsUsed", "tokensUsed" @@ -1439,8 +1496,10 @@ "type": "string", "enum": [ "processing", + "cancelling", "completed", - "failed" + "failed", + "cancelled" ] }, "results": { @@ -1448,7 +1507,7 @@ "items": { "$ref": "#/components/schemas/ExtractUrlResult" }, - "description": "Per-URL results in original request order." + "description": "Fixed-cardinality per-URL slots in original request order. Pending slots are processing; untouched terminal slots are cancelled." }, "error": { "type": "string", diff --git a/crates/crw-server/openapi/openapi.json b/crates/crw-server/openapi/openapi.json index aebc81aa..2a9ef712 100644 --- a/crates/crw-server/openapi/openapi.json +++ b/crates/crw-server/openapi/openapi.json @@ -777,8 +777,22 @@ "extract" ], "summary": "Start an async multi-URL structured extraction job", - "description": "Native extraction. Returns a per-URL array via GET /v1/extract/{id}. At least one of `prompt` or `schema` is required.", + "description": "Native extraction. Returns a fixed-cardinality per-URL array via GET /v1/extract/{id}. At least one of `prompt` or `schema` is required. SDK async starters send `Prefer: respond-async` for managed/self-hosted parity.", "operationId": "extractStart", + "parameters": [ + { + "name": "Prefer", + "in": "header", + "required": false, + "schema": { + "type": "string", + "enum": [ + "respond-async" + ] + }, + "description": "Request the canonical async lifecycle across managed and self-hosted deployments." + } + ], "requestBody": { "required": true, "content": { @@ -835,6 +849,46 @@ } } }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "404": { + "$ref": "#/components/responses/NotFound" + } + } + }, + "delete": { + "tags": [ + "extract" + ], + "summary": "Cancel an extract job", + "description": "Idempotent. Processing becomes cancelling while a claimed URL settles, then cancelled. Completed, failed, and cancelled states are immutable.", + "operationId": "extractCancel", + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "description": "Extract job id returned by POST /v1/extract" + } + ], + "responses": { + "200": { + "description": "Canonical job status + ordered per-URL results", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ExtractStatus" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, "404": { "$ref": "#/components/responses/NotFound" } @@ -2006,14 +2060,16 @@ "status": { "type": "string", "enum": [ + "processing", "completed", - "failed" + "failed", + "cancelled" ] }, "data": { "type": "object", "additionalProperties": true, - "description": "Extracted JSON object (present when status is completed)." + "description": "Extracted JSON object (present when status is completed). Untouched cancelled slots omit data, error, usage, and basis." }, "error": { "type": "string" @@ -2047,6 +2103,7 @@ "required": [ "id", "status", + "results", "expiresAt", "creditsUsed", "tokensUsed" @@ -2059,8 +2116,10 @@ "type": "string", "enum": [ "processing", + "cancelling", "completed", - "failed" + "failed", + "cancelled" ] }, "results": { @@ -2068,7 +2127,7 @@ "items": { "$ref": "#/components/schemas/ExtractUrlResult" }, - "description": "Per-URL results in original request order." + "description": "Fixed-cardinality per-URL slots in original request order. Pending slots are processing; untouched terminal slots are cancelled." }, "error": { "type": "string", diff --git a/crates/crw-server/src/routes/extract.rs b/crates/crw-server/src/routes/extract.rs index 112203ab..1f61e7d1 100644 --- a/crates/crw-server/src/routes/extract.rs +++ b/crates/crw-server/src/routes/extract.rs @@ -17,8 +17,8 @@ use crw_core::evidence::{Basis, BasisWarning}; use crw_core::types::{ExtractOptions, LlmUsage, OutputFormat, ScrapeRequest}; use crate::error::AppError; -use crate::routes::v2::adapters::expires_at_rfc3339; -use crate::state::{AppState, PreparedUrl, UrlResult}; +use crate::routes::v2::adapters::system_time_rfc3339; +use crate::state::{AppState, ExtractRecord, PreparedUrl, UrlResult}; /// Native extract request. camelCase like every other v1 public type. /// NOTE: no `#[derive(Debug)]` — `llm_api_key` is a secret and must never land @@ -268,18 +268,10 @@ pub struct ExtractStatusResponse { pub tokens_used: u32, } -pub async fn get_extract( - State(state): State, - Path(id): Path, -) -> Result, AppError> { - let rec = { - let jobs = state.extract_jobs.read().await; - jobs.get(&id) - .cloned() - .ok_or_else(|| CrwError::NotFound(format!("Extract job {id} not found")))? - }; - let expires_at = expires_at_rfc3339(rec.created_at, state.config.crawler.job_ttl_secs); - Ok(Json(ExtractStatusResponse { +/// The one canonical HTTP/MCP serializer for extract lifecycle state. +pub(crate) fn serialize_extract_status(id: Uuid, rec: ExtractRecord) -> ExtractStatusResponse { + let expires_at = system_time_rfc3339(rec.expires_at); + ExtractStatusResponse { id: id.to_string(), status: rec.status.as_str().to_string(), results: rec @@ -291,5 +283,35 @@ pub async fn get_extract( expires_at, credits_used: rec.credits_used, tokens_used: rec.tokens_used, - })) + } +} + +pub(crate) async fn get_extract_status( + state: &AppState, + id: Uuid, +) -> Result { + let rec = state.get_extract_job(id).await?; + Ok(serialize_extract_status(id, rec)) +} + +pub(crate) async fn cancel_extract_status( + state: &AppState, + id: Uuid, +) -> Result { + let rec = state.cancel_extract_job(id).await?; + Ok(serialize_extract_status(id, rec)) +} + +pub async fn get_extract( + State(state): State, + Path(id): Path, +) -> Result, AppError> { + Ok(Json(get_extract_status(&state, id).await?)) +} + +pub async fn cancel_extract( + State(state): State, + Path(id): Path, +) -> Result, AppError> { + Ok(Json(cancel_extract_status(&state, id).await?)) } diff --git a/crates/crw-server/src/routes/mcp.rs b/crates/crw-server/src/routes/mcp.rs index 14987e38..fb3c81ca 100644 --- a/crates/crw-server/src/routes/mcp.rs +++ b/crates/crw-server/src/routes/mcp.rs @@ -21,6 +21,20 @@ pub async fn validate_url(url: &str) -> Result<(), String> { crw_core::url_safety::validate_safe_url_resolved(&parsed).await } +/// Serialize an extract lifecycle response for MCP, re-adding the `success` +/// field MCP clients rely on. The shared HTTP `/v1` type omits it, so it is +/// injected only here; `success` is false only when every URL failed. +fn extract_status_value( + response: crate::routes::extract::ExtractStatusResponse, +) -> Result { + let success = response.status != "failed"; + let mut value = serde_json::to_value(response).map_err(|e| format!("serialize error: {e}"))?; + if let Some(obj) = value.as_object_mut() { + obj.insert("success".to_string(), Value::Bool(success)); + } + Ok(value) +} + pub async fn call_tool(state: &AppState, tool_name: &str, args: Value) -> Result { match tool_name { "crw_scrape" => { @@ -127,7 +141,7 @@ pub async fn call_tool(state: &AppState, tool_name: &str, args: Value) -> Result Ok(json!({"success": true, "id": id.to_string(), "status": "processing", "urls": urls})) } "crw_check_extract_status" => { - use crate::routes::extract::ExtractUrlResult; + use crate::routes::extract::get_extract_status; let id_str = args .get("id") .and_then(|v| v.as_str()) @@ -135,26 +149,24 @@ pub async fn call_tool(state: &AppState, tool_name: &str, args: Value) -> Result let id: Uuid = id_str .parse() .map_err(|_| format!("invalid extract id: {id_str}"))?; - let rec = { - let jobs = state.extract_jobs.read().await; - jobs.get(&id) - .cloned() - .ok_or(format!("extract job {id} not found"))? - }; - let results: Vec = rec - .per_url - .into_iter() - .map(ExtractUrlResult::from) - .collect(); - serde_json::to_value(json!({ - "success": !matches!(rec.status, crate::state::ExtractStatus::Failed), - "status": rec.status.as_str(), - "results": results, - "error": rec.error, - "creditsUsed": rec.credits_used, - "tokensUsed": rec.tokens_used, - })) - .map_err(|e| format!("serialize error: {e}")) + let response = get_extract_status(state, id) + .await + .map_err(|e| format!("{e}"))?; + extract_status_value(response) + } + "crw_cancel_extract" => { + use crate::routes::extract::cancel_extract_status; + let id_str = args + .get("id") + .and_then(|v| v.as_str()) + .ok_or("missing required parameter: id")?; + let id: Uuid = id_str + .parse() + .map_err(|_| format!("invalid extract id: {id_str}"))?; + let response = cancel_extract_status(state, id) + .await + .map_err(|e| format!("{e}"))?; + extract_status_value(response) } "crw_parse_file" => { use base64::Engine; diff --git a/crates/crw-server/src/routes/v1/mod.rs b/crates/crw-server/src/routes/v1/mod.rs index c79ca8e9..b8535c12 100644 --- a/crates/crw-server/src/routes/v1/mod.rs +++ b/crates/crw-server/src/routes/v1/mod.rs @@ -45,7 +45,9 @@ pub fn router() -> Router { ) .route( "/v1/extract/{id}", - get(routes::extract::get_extract).fallback(method_not_allowed), + get(routes::extract::get_extract) + .delete(routes::extract::cancel_extract) + .fallback(method_not_allowed), ) .route( "/v1/map", diff --git a/crates/crw-server/src/routes/v2/adapters.rs b/crates/crw-server/src/routes/v2/adapters.rs index 114a50f7..a403f4a4 100644 --- a/crates/crw-server/src/routes/v2/adapters.rs +++ b/crates/crw-server/src/routes/v2/adapters.rs @@ -234,6 +234,16 @@ pub fn expires_at_rfc3339(created_at: Instant, job_ttl_secs: u64) -> String { rfc3339_utc(now + remaining) } +/// Format a persisted absolute job expiry. Unlike `expires_at_rfc3339`, this +/// does not derive a new wall-clock value at read time. +pub fn system_time_rfc3339(expires_at: SystemTime) -> String { + let unix_secs = expires_at + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_secs()) + .unwrap_or(0); + rfc3339_utc(unix_secs) +} + /// Format a Unix-epoch second count as `YYYY-MM-DDTHH:MM:SS.000Z` (UTC). /// Hand-rolled (Howard Hinnant's `civil_from_days`) to avoid a chrono/time /// dependency. diff --git a/crates/crw-server/src/routes/v2/extract.rs b/crates/crw-server/src/routes/v2/extract.rs index d81ba825..0392add5 100644 --- a/crates/crw-server/src/routes/v2/extract.rs +++ b/crates/crw-server/src/routes/v2/extract.rs @@ -14,7 +14,7 @@ use uuid::Uuid; use crw_core::error::CrwError; use crw_core::types::{OutputFormat, ScrapeRequest}; -use super::adapters::expires_at_rfc3339; +use super::adapters::system_time_rfc3339; use crate::error::AppError; use crate::state::{AppState, ExtractStatus, PreparedUrl}; @@ -113,13 +113,8 @@ pub async fn get_extract( State(state): State, Path(id): Path, ) -> Result, AppError> { - let rec = { - let jobs = state.extract_jobs.read().await; - jobs.get(&id) - .cloned() - .ok_or_else(|| CrwError::NotFound(format!("Extract job {id} not found")))? - }; - let expires_at = expires_at_rfc3339(rec.created_at, state.config.crawler.job_ttl_secs); + let rec = state.get_extract_job(id).await?; + let expires_at = system_time_rfc3339(rec.expires_at); Ok(Json(V2ExtractStatusResponse { success: !matches!(rec.status, ExtractStatus::Failed), status: rec.status.as_str().to_string(), diff --git a/crates/crw-server/src/state.rs b/crates/crw-server/src/state.rs index 084288a1..26e26576 100644 --- a/crates/crw-server/src/state.rs +++ b/crates/crw-server/src/state.rs @@ -12,7 +12,7 @@ use crw_search::SearxngClient; use futures::stream::StreamExt; use std::collections::HashMap; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, SystemTime}; use tokio::sync::{RwLock, watch}; use uuid::Uuid; @@ -100,22 +100,33 @@ const MAX_CONCURRENT_CRAWLS: usize = 10; /// Interval between expired crawl job cleanup runs. const JOB_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); -/// Lifecycle of an async `/v2/extract` job. +/// Canonical lifecycle of an async extract job. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ExtractStatus { Processing, + Cancelling, Completed, Failed, + Cancelled, } impl ExtractStatus { pub fn as_str(self) -> &'static str { match self { ExtractStatus::Processing => "processing", + ExtractStatus::Cancelling => "cancelling", ExtractStatus::Completed => "completed", ExtractStatus::Failed => "failed", + ExtractStatus::Cancelled => "cancelled", } } + + pub fn is_terminal(self) -> bool { + matches!( + self, + ExtractStatus::Completed | ExtractStatus::Failed | ExtractStatus::Cancelled + ) + } } /// One URL's extraction outcome. Powers the native `/v1/extract` per-URL array @@ -161,6 +172,88 @@ pub struct ExtractRecord { pub credits_used: u32, pub error: Option, pub created_at: Instant, + /// Absolute wall-clock expiry captured once at admission. Serializers use + /// this persisted value so repeated lifecycle envelopes never drift. + pub expires_at: SystemTime, + /// The one URL currently dispatched by the sequential worker. Cancellation + /// cannot cross its terminal barrier until this slot has persisted. + pub claimed_index: Option, +} + +impl ExtractRecord { + fn is_expired(&self, ttl: Duration) -> bool { + self.created_at.elapsed() >= ttl + } + + /// Complete the cancellation barrier. Call only while holding the extract + /// jobs write lock and after observing that no URL remains claimed. + fn finish_cancellation(&mut self) { + if self.status != ExtractStatus::Cancelling || self.claimed_index.is_some() { + return; + } + let mut cancelled_any = false; + for result in &mut self.per_url { + if result.status == ExtractStatus::Processing { + result.status = ExtractStatus::Cancelled; + result.data = None; + result.error = None; + result.llm_usage = None; + result.basis = None; + result.basis_warnings.clear(); + result.llm_input_hash = None; + cancelled_any = true; + } + } + if cancelled_any { + // A genuine cancel: at least one in-flight URL was actually stopped. + self.status = ExtractStatus::Cancelled; + } else { + // Every URL had already reached a terminal state before the cancel + // landed. Reporting "cancelled" would contradict the per-URL results + // (all completed/failed with real data), so settle it as the + // naturally finished job it actually is. + self.complete_from_outcomes(); + } + } + + /// Set the terminal job status from the per-URL outcomes of a job that ran + /// to the end: completed if any URL succeeded, otherwise failed, with the + /// one-credit floor a naturally finished job carries. + fn complete_from_outcomes(&mut self) { + let any_ok = self + .per_url + .iter() + .any(|result| result.status == ExtractStatus::Completed); + if any_ok { + self.status = ExtractStatus::Completed; + self.data + .get_or_insert_with(|| serde_json::Value::Object(Default::default())); + } else { + self.status = ExtractStatus::Failed; + self.error = self + .per_url + .iter() + .rev() + .find_map(|result| result.error.clone()); + } + // Preserve the existing one-credit floor for a naturally finished + // all-failed job. Cancelled jobs retain only measured usage. + self.credits_used = self.credits_used.max(1); + } + + /// Commit the worker's final job-level write. DELETE races this method on + /// the same map lock, so exactly one transition can win and terminal state + /// is never rewritten by the loser. + fn finish_processing(&mut self) { + if self.status != ExtractStatus::Processing { + if self.status == ExtractStatus::Cancelling { + self.finish_cancellation(); + } + return; + } + + self.complete_from_outcomes(); + } } /// Shared application state. @@ -333,12 +426,9 @@ impl AppState { } drop(jobs); - // Prune finished extract jobs past TTL (keep in-flight ones). - let mut ejobs = cleanup_state.extract_jobs.write().await; - ejobs.retain(|_id, rec| { - matches!(rec.status, ExtractStatus::Processing) - || rec.created_at.elapsed() < ttl - }); + // TTL is authoritative for every extract lifecycle state. A + // stalled processing/cancelling job must not live forever. + cleanup_state.prune_expired_extract_jobs(ttl).await; } }); @@ -617,18 +707,45 @@ impl AppState { template: ScrapeRequest, ) -> Uuid { let id = Uuid::new_v4(); + // Seed the fixed-cardinality result array before the worker can run. + // Preflight failures are already final; valid URLs remain processing + // until claimed and persisted, or are converted to cancelled at the + // cancellation barrier. + let per_url = entries + .iter() + .map(|entry| UrlResult { + url: entry.url.clone(), + status: if entry.preflight_error.is_some() { + ExtractStatus::Failed + } else { + ExtractStatus::Processing + }, + data: None, + error: entry.preflight_error.clone(), + llm_usage: None, + basis: None, + basis_warnings: Vec::new(), + llm_input_hash: None, + }) + .collect(); { let mut jobs = self.extract_jobs.write().await; + let created_at = Instant::now(); + let wall_now = SystemTime::now(); + let expires_at = + wall_now.checked_add(Duration::from_secs(self.config.crawler.job_ttl_secs)); jobs.insert( id, ExtractRecord { status: ExtractStatus::Processing, data: None, - per_url: Vec::new(), + per_url, tokens_used: 0, credits_used: 0, error: None, - created_at: Instant::now(), + created_at, + expires_at: expires_at.unwrap_or(wall_now), + claimed_index: None, }, ); } @@ -636,6 +753,7 @@ impl AppState { let renderer = self.renderer.clone(); let config = self.config.clone(); let extract_jobs = self.extract_jobs.clone(); + let finalizer = self.clone(); tokio::spawn(async move { // `/v2/extract` is a multi-URL background job — `Batch` traffic, so its @@ -651,35 +769,42 @@ impl AppState { let deadline_ms = config.effective_deadline_ms(template.deadline_ms, template.wait_for); - let mut merged = serde_json::Map::new(); - let mut per_url: Vec = Vec::with_capacity(entries.len()); - let mut tokens = 0u32; - let mut credits = 0u32; - let mut last_err: Option = None; - let mut any_ok = false; - - for entry in entries { - // Preflight-failed URLs (bad parse / SSRF) surface as - // `failed` without a fetch — never silently dropped. - if let Some(err) = entry.preflight_error { - last_err = Some(err.clone()); - per_url.push(UrlResult { - url: entry.url, - status: ExtractStatus::Failed, - data: None, - error: Some(err), - llm_usage: None, - basis: None, - basis_warnings: Vec::new(), - llm_input_hash: None, - }); + for (index, entry) in entries.into_iter().enumerate() { + // Preflight failures were persisted at admission and + // never count as dispatched work. + if entry.preflight_error.is_some() { continue; } + // Claim exactly one slot while holding the same state + // lock DELETE uses. Once cancelling is visible no new + // URL can cross this point. + { + let mut jobs = extract_jobs.write().await; + let Some(rec) = jobs.get_mut(&id) else { + return; + }; + match rec.status { + ExtractStatus::Processing => { + if rec.per_url[index].status != ExtractStatus::Processing { + continue; + } + rec.claimed_index = Some(index); + } + ExtractStatus::Cancelling => { + rec.finish_cancellation(); + return; + } + ExtractStatus::Completed + | ExtractStatus::Failed + | ExtractStatus::Cancelled => return, + } + } + let mut req = template.clone(); req.url = entry.url.clone(); let deadline = Deadline::from_request_ms(deadline_ms); - match scrape_url( + let (result, merged_fields, tokens, credits) = match scrape_url( &req, &renderer, config.extraction.llm.as_ref(), @@ -692,64 +817,286 @@ impl AppState { .await { Ok(d) => { - any_ok = true; - if let Some(serde_json::Value::Object(obj)) = &d.json { - for (k, v) in obj { - merged.insert(k.clone(), v.clone()); - } - } - if let Some(usage) = &d.llm_usage { - tokens += usage.total_tokens; - } - credits += if d.credit_cost == 0 { 1 } else { d.credit_cost }; - per_url.push(UrlResult { - url: entry.url, - status: ExtractStatus::Completed, - data: d.json, - error: None, - llm_usage: d.llm_usage, - basis: d.basis, - basis_warnings: d.basis_warnings, - llm_input_hash: d.llm_input_hash, - }); + let merged_fields = match &d.json { + Some(serde_json::Value::Object(obj)) => Some(obj.clone()), + _ => None, + }; + let tokens = + d.llm_usage.as_ref().map_or(0, |usage| usage.total_tokens); + let credits = if d.credit_cost == 0 { 1 } else { d.credit_cost }; + ( + UrlResult { + url: entry.url, + status: ExtractStatus::Completed, + data: d.json, + error: None, + llm_usage: d.llm_usage, + basis: d.basis, + basis_warnings: d.basis_warnings, + llm_input_hash: d.llm_input_hash, + }, + merged_fields, + tokens, + credits, + ) } Err(e) => { let msg = e.to_string(); - last_err = Some(msg.clone()); - per_url.push(UrlResult { - url: entry.url, - status: ExtractStatus::Failed, - data: None, - error: Some(msg), - llm_usage: None, - basis: None, - basis_warnings: Vec::new(), - llm_input_hash: None, - }); + ( + UrlResult { + url: entry.url, + status: ExtractStatus::Failed, + data: None, + error: Some(msg), + llm_usage: None, + basis: None, + basis_warnings: Vec::new(), + llm_input_hash: None, + }, + None, + 0, + 0, + ) } - } - } + }; - let mut jobs = extract_jobs.write().await; - if let Some(rec) = jobs.get_mut(&id) { - if !any_ok && last_err.is_some() { - rec.status = ExtractStatus::Failed; - rec.error = last_err; - } else { - rec.status = ExtractStatus::Completed; - rec.data = Some(serde_json::Value::Object(merged)); + // Persist the completed claimed slot and cumulative + // measured usage before dispatching anything else. + let mut jobs = extract_jobs.write().await; + let Some(rec) = jobs.get_mut(&id) else { + return; + }; + if rec.status.is_terminal() || rec.claimed_index != Some(index) { + return; + } + rec.per_url[index] = result; + if let Some(fields) = merged_fields { + let merged = rec.data.get_or_insert_with(|| { + serde_json::Value::Object(Default::default()) + }); + if let serde_json::Value::Object(merged) = merged { + merged.extend(fields); + } + } + rec.tokens_used = rec.tokens_used.saturating_add(tokens); + rec.credits_used = rec.credits_used.saturating_add(credits); + rec.claimed_index = None; + if rec.status == ExtractStatus::Cancelling { + rec.finish_cancellation(); + return; } - rec.per_url = per_url; - rec.tokens_used = tokens; - // ponytail: 1-credit floor even on an all-failed job — - // preflight-failed URLs add 0 (they `continue` before the - // tally). SaaS settles the real cost separately. - rec.credits_used = credits.max(1); } + + finalizer.finalize_extract_job(id).await; }) .await; }); id } + + /// Request cancellation and return the persisted canonical state. Repeated + /// calls are idempotent; terminal jobs are never rewritten. + pub async fn cancel_extract_job(&self, id: Uuid) -> CrwResult { + let mut jobs = self.extract_jobs.write().await; + let ttl = Duration::from_secs(self.config.crawler.job_ttl_secs); + if jobs.get(&id).is_some_and(|rec| rec.is_expired(ttl)) { + jobs.remove(&id); + return Err(CrwError::NotFound(format!("Extract job {id} not found"))); + } + let rec = jobs + .get_mut(&id) + .ok_or_else(|| CrwError::NotFound(format!("Extract job {id} not found")))?; + if rec.status == ExtractStatus::Processing { + rec.status = ExtractStatus::Cancelling; + } + if rec.status == ExtractStatus::Cancelling { + rec.finish_cancellation(); + } + Ok(rec.clone()) + } + + /// TTL-aware canonical lookup shared by v1, v2, and MCP handlers. A write + /// lock makes expiry observation and removal one atomic operation. + pub async fn get_extract_job(&self, id: Uuid) -> CrwResult { + let mut jobs = self.extract_jobs.write().await; + let ttl = Duration::from_secs(self.config.crawler.job_ttl_secs); + if jobs.get(&id).is_some_and(|rec| rec.is_expired(ttl)) { + jobs.remove(&id); + return Err(CrwError::NotFound(format!("Extract job {id} not found"))); + } + jobs.get(&id) + .cloned() + .ok_or_else(|| CrwError::NotFound(format!("Extract job {id} not found"))) + } + + async fn finalize_extract_job(&self, id: Uuid) { + let mut jobs = self.extract_jobs.write().await; + if let Some(rec) = jobs.get_mut(&id) { + rec.finish_processing(); + } + } + + async fn prune_expired_extract_jobs(&self, ttl: Duration) -> usize { + let mut jobs = self.extract_jobs.write().await; + let before = jobs.len(); + jobs.retain(|_id, rec| !rec.is_expired(ttl)); + before - jobs.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::routes::extract::serialize_extract_status; + use serde_json::json; + use tokio::sync::oneshot; + + fn completed_record(created_at: Instant) -> ExtractRecord { + ExtractRecord { + status: ExtractStatus::Processing, + data: Some(json!({"last": 3})), + per_url: (1..=3) + .map(|index| UrlResult { + url: format!("https://example.com/{index}"), + status: ExtractStatus::Completed, + data: Some(json!({"index": index})), + error: None, + llm_usage: None, + basis: None, + basis_warnings: Vec::new(), + llm_input_hash: None, + }) + .collect(), + tokens_used: 30, + credits_used: 3, + error: None, + created_at, + expires_at: SystemTime::now() + Duration::from_secs(3_600), + claimed_index: None, + } + } + + fn spawn_final_write( + state: AppState, + id: Uuid, + ) -> (oneshot::Receiver<()>, tokio::task::JoinHandle<()>) { + let (ready_tx, ready_rx) = oneshot::channel(); + let handle = tokio::spawn(async move { + let _ = ready_tx.send(()); + state.finalize_extract_job(id).await; + }); + (ready_rx, handle) + } + + fn spawn_delete( + state: AppState, + id: Uuid, + ) -> (oneshot::Receiver<()>, tokio::task::JoinHandle<()>) { + let (ready_tx, ready_rx) = oneshot::channel(); + let handle = tokio::spawn(async move { + let _ = ready_tx.send(()); + state.cancel_extract_job(id).await.unwrap(); + }); + (ready_rx, handle) + } + + async fn run_final_write_delete_race(final_write_first: bool) -> (AppState, Uuid) { + let config: AppConfig = toml::from_str("").unwrap(); + let state = AppState::new(config).unwrap(); + let id = Uuid::new_v4(); + state + .extract_jobs + .write() + .await + .insert(id, completed_record(Instant::now())); + + // Hold the exact lock used by both operations, then enqueue them in a + // known order. Tokio's fair write lock releases to the first waiter, + // making both legal race winners deterministic rather than probabilistic. + let gate = state.extract_jobs.write().await; + let ((first_ready, first), (second_ready, second)) = if final_write_first { + ( + spawn_final_write(state.clone(), id), + spawn_delete(state.clone(), id), + ) + } else { + ( + spawn_delete(state.clone(), id), + spawn_final_write(state.clone(), id), + ) + }; + first_ready.await.unwrap(); + second_ready.await.unwrap(); + drop(gate); + first.await.unwrap(); + second.await.unwrap(); + (state, id) + } + + #[tokio::test] + async fn final_write_versus_delete_race_covers_both_winners_and_freezes_terminal_state() { + // Whichever of the final write or the DELETE wins, a job whose URLs all + // completed settles as Completed: a late cancel that stopped nothing + // in-flight must not relabel a finished job (the per-URL results below + // are all Completed in both branches). + for (final_write_first, expected) in [ + (true, ExtractStatus::Completed), + (false, ExtractStatus::Completed), + ] { + let (state, id) = run_final_write_delete_race(final_write_first).await; + let terminal = state.get_extract_job(id).await.unwrap(); + assert_eq!(terminal.status, expected); + assert_eq!(terminal.per_url.len(), 3); + assert_eq!( + terminal + .per_url + .iter() + .map(|result| result.url.as_str()) + .collect::>(), + [ + "https://example.com/1", + "https://example.com/2", + "https://example.com/3" + ] + ); + assert!( + terminal + .per_url + .iter() + .all(|result| result.status == ExtractStatus::Completed) + ); + + let frozen = serde_json::to_value(serialize_extract_status(id, terminal)).unwrap(); + state.finalize_extract_job(id).await; + let repeated_delete = state.cancel_extract_job(id).await.unwrap(); + let repeated = + serde_json::to_value(serialize_extract_status(id, repeated_delete)).unwrap(); + assert_eq!(repeated, frozen, "terminal envelope must be immutable"); + } + } + + #[tokio::test] + async fn cleanup_removes_expired_nonterminal_extract_jobs() { + let config: AppConfig = toml::from_str("").unwrap(); + let state = AppState::new(config).unwrap(); + let processing_id = Uuid::new_v4(); + let cancelling_id = Uuid::new_v4(); + let old = Instant::now() - Duration::from_secs(5); + let mut cancelling = completed_record(old); + cancelling.status = ExtractStatus::Cancelling; + { + let mut jobs = state.extract_jobs.write().await; + jobs.insert(processing_id, completed_record(old)); + jobs.insert(cancelling_id, cancelling); + } + + assert_eq!( + state + .prune_expired_extract_jobs(Duration::from_secs(1)) + .await, + 2 + ); + assert!(state.extract_jobs.read().await.is_empty()); + } } diff --git a/crates/crw-server/tests/mcp.rs b/crates/crw-server/tests/mcp.rs index 45cd71f3..5a07d237 100644 --- a/crates/crw-server/tests/mcp.rs +++ b/crates/crw-server/tests/mcp.rs @@ -6,8 +6,10 @@ use crw_core::types::{ ApiResponse, GroupedSearchData, ImageResult, SearchData, SearchResponseData, SearchResult, }; use crw_server::app::create_app; -use crw_server::state::AppState; +use crw_server::state::{AppState, ExtractRecord, ExtractStatus, UrlResult}; use serde_json::{Value, json}; +use std::time::{Duration, Instant, SystemTime}; +use uuid::Uuid; fn test_app() -> TestServer { let config: AppConfig = toml::from_str("").unwrap(); @@ -28,6 +30,29 @@ fn test_app_with_search() -> TestServer { TestServer::new(app) } +fn pending_extract_record(created_at: Instant) -> ExtractRecord { + ExtractRecord { + status: ExtractStatus::Processing, + data: None, + per_url: vec![UrlResult { + url: "https://example.com".into(), + status: ExtractStatus::Processing, + data: None, + error: None, + llm_usage: None, + basis: None, + basis_warnings: Vec::new(), + llm_input_hash: None, + }], + tokens_used: 0, + credits_used: 0, + error: None, + created_at, + expires_at: SystemTime::now() + Duration::from_secs(3_600), + claimed_index: None, + } +} + fn mcp_request( method: &str, id: serde_json::Value, @@ -64,7 +89,7 @@ async fn mcp_initialize_returns_capabilities() { #[tokio::test] async fn mcp_tools_list_returns_all_tools() { - // With a search backend configured, all 8 tools are advertised. + // With a search backend configured, all 9 tools are advertised. let server = test_app_with_search(); let resp = server .post("/mcp") @@ -76,8 +101,8 @@ async fn mcp_tools_list_returns_all_tools() { let tools = json["result"]["tools"].as_array().unwrap(); assert_eq!( tools.len(), - 8, - "Should have 8 tools: scrape, crawl, check, map, search, parse_file, extract, check_extract" + 9, + "Should have 9 tools including extract start/status/cancel" ); let tool_names: Vec<&str> = tools.iter().map(|t| t["name"].as_str().unwrap()).collect(); @@ -89,10 +114,11 @@ async fn mcp_tools_list_returns_all_tools() { assert!(tool_names.contains(&"crw_parse_file")); assert!(tool_names.contains(&"crw_extract")); assert!(tool_names.contains(&"crw_check_extract_status")); + assert!(tool_names.contains(&"crw_cancel_extract")); // Every tool advertises annotations. Job-starting tools (crw_crawl, // crw_extract) have side effects and must NOT be marked read-only. - let non_read_only = ["crw_crawl", "crw_extract"]; + let non_read_only = ["crw_crawl", "crw_extract", "crw_cancel_extract"]; for t in tools { assert!( t["annotations"].is_object(), @@ -115,6 +141,20 @@ async fn mcp_tools_list_returns_all_tools() { assert_eq!(crawl["annotations"]["idempotentHint"], false); let scrape = tools.iter().find(|t| t["name"] == "crw_scrape").unwrap(); assert_eq!(scrape["annotations"]["readOnlyHint"], true); + let cancel = tools + .iter() + .find(|t| t["name"] == "crw_cancel_extract") + .unwrap(); + assert_eq!(cancel["annotations"]["destructiveHint"], true); + assert_eq!(cancel["annotations"]["idempotentHint"], true); + for name in [ + "crw_extract", + "crw_check_extract_status", + "crw_cancel_extract", + ] { + let tool = tools.iter().find(|t| t["name"] == name).unwrap(); + assert!(tool["outputSchema"].is_object(), "{name} outputSchema"); + } } /// crw_search is suppressed from tools/list when no search backend is configured, @@ -131,11 +171,199 @@ async fn mcp_tools_list_hides_search_without_backend() { let json: serde_json::Value = resp.json(); let tools = json["result"]["tools"].as_array().unwrap(); let names: Vec<&str> = tools.iter().map(|t| t["name"].as_str().unwrap()).collect(); - assert_eq!(tools.len(), 7, "crw_search hidden without a backend"); + assert_eq!(tools.len(), 8, "crw_search hidden without a backend"); assert!(!names.contains(&"crw_search")); assert!(names.contains(&"crw_scrape")); } +#[tokio::test] +async fn mcp_extract_status_and_cancel_share_canonical_http_serializer() { + let config: AppConfig = toml::from_str("").unwrap(); + let state = AppState::new(config).unwrap(); + let id = Uuid::new_v4(); + state + .extract_jobs + .write() + .await + .insert(id, pending_extract_record(Instant::now())); + let server = TestServer::new(create_app(state)); + + let status: Value = server + .post("/mcp") + .content_type("application/json") + .json(&mcp_request( + "tools/call", + json!(201), + json!({"name": "crw_check_extract_status", "arguments": {"id": id}}), + )) + .await + .json(); + let status = &status["result"]["structuredContent"]; + assert_eq!(status["id"], id.to_string()); + assert_eq!(status["status"], "processing"); + assert_eq!(status["results"][0]["status"], "processing"); + assert!(status["expiresAt"].is_string()); + // MCP keeps the `success` envelope every other MCP tool returns; it is false + // only for a job whose every URL failed (here the job is still processing). + assert_eq!(status["success"], true); + let schema = tool_output_schema("crw_check_extract_status").unwrap(); + let validator = jsonschema::validator_for(&schema).expect("extract schema compiles"); + let errors: Vec = validator + .iter_errors(status) + .map(|error| error.to_string()) + .collect(); + assert!(errors.is_empty(), "serializer/schema drift: {errors:#?}"); + + let cancelled: Value = server + .post("/mcp") + .content_type("application/json") + .json(&mcp_request( + "tools/call", + json!(202), + json!({"name": "crw_cancel_extract", "arguments": {"id": id}}), + )) + .await + .json(); + let cancelled = &cancelled["result"]["structuredContent"]; + assert_eq!(cancelled["status"], "cancelled"); + assert_eq!(cancelled["results"][0]["status"], "cancelled"); + // A cancelled job is not a failed one, so it still reports success: true. + assert_eq!(cancelled["success"], true); +} + +#[tokio::test] +async fn mcp_extract_output_schemas_match_openapi_lifecycle_components() { + let openapi: Value = + serde_json::from_str(include_str!("../openapi/openapi.json")).expect("valid OpenAPI"); + let components = &openapi["components"]["schemas"]; + + // MCP responses carry the `success` envelope every MCP tool returns, which + // the native HTTP `/v1` (openapi) shape deliberately omits. Compare the data + // contract by ignoring that one envelope field on the MCP side. + let required_without_success = |schema: &Value| -> Vec { + schema["required"] + .as_array() + .unwrap() + .iter() + .filter_map(|v| v.as_str()) + .filter(|s| *s != "success") + .map(str::to_string) + .collect() + }; + let keys_without_success = |schema: &Value| -> Vec { + schema["properties"] + .as_object() + .unwrap() + .keys() + .filter(|k| *k != "success") + .cloned() + .collect() + }; + + let accepted = tool_output_schema("crw_extract").unwrap(); + let openapi_accepted = &components["ExtractAccepted"]; + assert_eq!( + required_without_success(&accepted), + required_without_success(openapi_accepted) + ); + assert_eq!( + accepted["properties"]["status"], + openapi_accepted["properties"]["status"] + ); + assert_eq!( + keys_without_success(&accepted), + keys_without_success(openapi_accepted) + ); + + let status = tool_output_schema("crw_check_extract_status").unwrap(); + assert_eq!(status, tool_output_schema("crw_cancel_extract").unwrap()); + let openapi_status = &components["ExtractStatus"]; + assert_eq!( + required_without_success(&status), + required_without_success(openapi_status) + ); + assert_eq!( + status["properties"]["status"], + openapi_status["properties"]["status"] + ); + + let result = &status["properties"]["results"]["items"]; + let openapi_result = &components["ExtractUrlResult"]; + assert_eq!(result["required"], openapi_result["required"]); + assert_eq!( + result["properties"] + .as_object() + .unwrap() + .keys() + .collect::>(), + openapi_result["properties"] + .as_object() + .unwrap() + .keys() + .collect::>() + ); + for field in [ + "url", + "status", + "data", + "error", + "llmUsage", + "basis", + "basisWarnings", + "llmInputHash", + ] { + assert_eq!( + result["properties"][field]["type"], openapi_result["properties"][field]["type"], + "type drift for ExtractUrlResult.{field}" + ); + } + assert_eq!( + result["properties"]["status"], + openapi_result["properties"]["status"] + ); +} + +#[tokio::test] +async fn mcp_extract_status_and_cancel_treat_expired_ids_as_not_found() { + let mut config: AppConfig = toml::from_str("").unwrap(); + config.crawler.job_ttl_secs = 1; + let state = AppState::new(config).unwrap(); + let ids = [Uuid::new_v4(), Uuid::new_v4()]; + let expired = pending_extract_record(Instant::now() - Duration::from_secs(2)); + { + let mut jobs = state.extract_jobs.write().await; + jobs.insert(ids[0], expired.clone()); + jobs.insert(ids[1], expired); + } + let server = TestServer::new(create_app(state.clone())); + + for (tool, id) in [ + ("crw_check_extract_status", ids[0]), + ("crw_cancel_extract", ids[1]), + ] { + let response: Value = server + .post("/mcp") + .content_type("application/json") + .json(&mcp_request( + "tools/call", + json!(301), + json!({"name": tool, "arguments": {"id": id}}), + )) + .await + .json(); + assert_eq!(response["result"]["isError"], true, "{tool}"); + assert!( + response["result"]["content"][0]["text"] + .as_str() + .unwrap() + .contains("not found"), + "{tool}" + ); + assert!(response["result"].get("structuredContent").is_none()); + } + assert!(state.extract_jobs.read().await.is_empty()); +} + #[tokio::test] async fn mcp_unknown_method_error_32601() { let server = test_app(); diff --git a/crates/crw-server/tests/v1_extract.rs b/crates/crw-server/tests/v1_extract.rs index 580e9ec8..0e2c280a 100644 --- a/crates/crw-server/tests/v1_extract.rs +++ b/crates/crw-server/tests/v1_extract.rs @@ -9,8 +9,10 @@ use axum::http::StatusCode; use axum_test::TestServer; use crw_core::config::AppConfig; use crw_server::app::create_app; -use crw_server::state::{AppState, ExtractStatus, PreparedUrl}; +use crw_server::state::{AppState, ExtractRecord, ExtractStatus, PreparedUrl, UrlResult}; use serde_json::{Value, json}; +use std::time::{Duration, Instant, SystemTime}; +use uuid::Uuid; fn test_app() -> TestServer { let config: AppConfig = toml::from_str("").unwrap(); @@ -117,6 +119,205 @@ async fn v1_extract_status_unknown_404() { r.assert_status(StatusCode::NOT_FOUND); } +#[tokio::test] +async fn v1_extract_cancel_rejects_malformed_and_unknown_ids() { + let s = test_app(); + s.delete("/v1/extract/not-a-uuid") + .await + .assert_status(StatusCode::BAD_REQUEST); + s.delete("/v1/extract/00000000-0000-0000-0000-000000000000") + .await + .assert_status(StatusCode::NOT_FOUND); +} + +#[tokio::test] +async fn expired_extract_ids_are_immediate_404_for_v1_get_delete_and_v2_get() { + let mut config: AppConfig = toml::from_str("").unwrap(); + config.crawler.job_ttl_secs = 1; + let state = AppState::new(config).unwrap(); + let ids = [Uuid::new_v4(), Uuid::new_v4(), Uuid::new_v4()]; + let mut expired = seeded_record(&["https://expired.example"], None); + expired.created_at = Instant::now() - Duration::from_secs(2); + { + let mut jobs = state.extract_jobs.write().await; + for id in ids { + jobs.insert(id, expired.clone()); + } + } + let server = TestServer::new(create_app(state.clone())); + + server + .get(&format!("/v1/extract/{}", ids[0])) + .await + .assert_status(StatusCode::NOT_FOUND); + server + .delete(&format!("/v1/extract/{}", ids[1])) + .await + .assert_status(StatusCode::NOT_FOUND); + server + .get(&format!("/v2/extract/{}", ids[2])) + .await + .assert_status(StatusCode::NOT_FOUND); + assert!(state.extract_jobs.read().await.is_empty()); +} + +fn pending_slot(url: &str) -> UrlResult { + UrlResult { + url: url.into(), + status: ExtractStatus::Processing, + data: None, + error: None, + llm_usage: None, + basis: None, + basis_warnings: Vec::new(), + llm_input_hash: None, + } +} + +fn seeded_record(urls: &[&str], claimed_index: Option) -> ExtractRecord { + ExtractRecord { + status: ExtractStatus::Processing, + data: None, + per_url: urls.iter().map(|url| pending_slot(url)).collect(), + tokens_used: 0, + credits_used: 0, + error: None, + created_at: Instant::now(), + expires_at: SystemTime::now() + Duration::from_secs(3_600), + claimed_index, + } +} + +#[tokio::test] +async fn cancel_before_dispatch_is_terminal_ordered_and_idempotent() { + let config: AppConfig = toml::from_str("").unwrap(); + let state = AppState::new(config).unwrap(); + let id = Uuid::new_v4(); + state.extract_jobs.write().await.insert( + id, + seeded_record(&["https://a.example", "https://b.example"], None), + ); + let server = TestServer::new(create_app(state)); + + let first: Value = server.delete(&format!("/v1/extract/{id}")).await.json(); + assert_eq!(first["status"], "cancelled"); + assert_eq!(first["results"].as_array().unwrap().len(), 2); + assert_eq!(first["results"][0]["url"], "https://a.example"); + assert_eq!(first["results"][1]["url"], "https://b.example"); + for result in first["results"].as_array().unwrap() { + assert_eq!(result["status"], "cancelled"); + assert_eq!(result.as_object().unwrap().len(), 2); + } + + let repeated: Value = server.delete(&format!("/v1/extract/{id}")).await.json(); + assert_eq!(repeated, first, "repeated DELETE returns persisted state"); + let get: Value = server.get(&format!("/v1/extract/{id}")).await.json(); + assert_eq!(get, first, "GET and DELETE share the canonical serializer"); +} + +#[tokio::test] +async fn terminal_envelope_uses_the_persisted_expiry_without_recomputation() { + let config: AppConfig = toml::from_str("").unwrap(); + let state = AppState::new(config).unwrap(); + let id = Uuid::new_v4(); + let mut record = seeded_record(&["https://done.example"], None); + record.status = ExtractStatus::Completed; + record.per_url[0].status = ExtractStatus::Completed; + record.per_url[0].data = Some(json!({"done": true})); + state.extract_jobs.write().await.insert(id, record); + let server = TestServer::new(create_app(state.clone())); + + let first: Value = server.get(&format!("/v1/extract/{id}")).await.json(); + // Move the monotonic admission time without changing the persisted wall + // expiry. The old read-time derivation would shift expiresAt by 10 seconds. + state + .extract_jobs + .write() + .await + .get_mut(&id) + .unwrap() + .created_at = Instant::now() - Duration::from_secs(10); + let repeated: Value = server.delete(&format!("/v1/extract/{id}")).await.json(); + assert_eq!(repeated, first); +} + +#[tokio::test] +async fn claimed_slot_holds_cancelling_barrier_until_its_result_persists() { + let config: AppConfig = toml::from_str("").unwrap(); + let state = AppState::new(config).unwrap(); + let id = Uuid::new_v4(); + state.extract_jobs.write().await.insert( + id, + seeded_record( + &["https://claimed.example", "https://pending.example"], + Some(0), + ), + ); + let server = TestServer::new(create_app(state.clone())); + + let cancelling: Value = server.delete(&format!("/v1/extract/{id}")).await.json(); + assert_eq!(cancelling["status"], "cancelling"); + assert_eq!(cancelling["results"][0]["status"], "processing"); + assert_eq!(cancelling["results"][1]["status"], "processing"); + let repeated: Value = server.delete(&format!("/v1/extract/{id}")).await.json(); + assert_eq!(repeated, cancelling); + + // Model the worker's atomic final write for the already claimed URL. + { + let mut jobs = state.extract_jobs.write().await; + let rec = jobs.get_mut(&id).unwrap(); + rec.per_url[0].status = ExtractStatus::Completed; + rec.per_url[0].data = Some(json!({"title": "persisted"})); + rec.tokens_used = 9; + rec.credits_used = 1; + rec.claimed_index = None; + } + + let terminal: Value = server.delete(&format!("/v1/extract/{id}")).await.json(); + assert_eq!(terminal["status"], "cancelled"); + assert_eq!(terminal["results"][0]["status"], "completed"); + assert_eq!(terminal["results"][0]["data"]["title"], "persisted"); + assert_eq!(terminal["results"][1]["status"], "cancelled"); + assert_eq!(terminal["tokensUsed"], 9); + assert_eq!(terminal["creditsUsed"], 1); +} + +#[tokio::test] +async fn delete_never_rewrites_completed_or_failed_terminal_state() { + let config: AppConfig = toml::from_str("").unwrap(); + let state = AppState::new(config).unwrap(); + let completed_id = Uuid::new_v4(); + let failed_id = Uuid::new_v4(); + let mut completed = seeded_record(&["https://done.example"], None); + completed.status = ExtractStatus::Completed; + completed.per_url[0].status = ExtractStatus::Completed; + completed.per_url[0].data = Some(json!({"ok": true})); + let mut failed = seeded_record(&["https://failed.example"], None); + failed.status = ExtractStatus::Failed; + failed.per_url[0].status = ExtractStatus::Failed; + failed.per_url[0].error = Some("failed".into()); + failed.error = Some("failed".into()); + { + let mut jobs = state.extract_jobs.write().await; + jobs.insert(completed_id, completed); + jobs.insert(failed_id, failed); + } + let server = TestServer::new(create_app(state)); + + let completed: Value = server + .delete(&format!("/v1/extract/{completed_id}")) + .await + .json(); + assert_eq!(completed["status"], "completed"); + assert_eq!(completed["results"][0]["data"]["ok"], true); + let failed: Value = server + .delete(&format!("/v1/extract/{failed_id}")) + .await + .json(); + assert_eq!(failed["status"], "failed"); + assert_eq!(failed["error"], "failed"); +} + #[tokio::test] async fn worker_seeds_preflight_failures_in_order() { // The worker processes entries in original order and surfaces preflight @@ -179,6 +380,136 @@ async fn worker_seeds_preflight_failures_in_order() { assert_eq!(rec.per_url[0].error.as_deref(), Some("bad-1")); } +#[tokio::test] +async fn worker_cancel_after_persisted_result_settles_claim_and_stops_next_dispatch() { + use crw_core::types::{OutputFormat, ScrapeRequest}; + use std::time::Duration; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + unsafe { + std::env::set_var("CRW_ALLOW_LOOPBACK_FOR_TESTS", "1"); + } + let mock = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/page-a")) + .respond_with( + ResponseTemplate::new(200).set_body_string("

A

"), + ) + .mount(&mock) + .await; + Mock::given(method("GET")) + .and(path("/page-b")) + .respond_with( + ResponseTemplate::new(200) + .set_body_string("

B

") + .set_delay(Duration::from_millis(75)), + ) + .mount(&mock) + .await; + Mock::given(method("GET")) + .and(path("/page-c")) + .respond_with(ResponseTemplate::new(200).set_body_string("

C

")) + .mount(&mock) + .await; + Mock::given(method("POST")) + .and(path("/v1/messages")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "content": [{ + "type": "tool_use", + "id": "t1", + "name": "extract_data", + "input": { "title": "A" } + }], + "usage": { "input_tokens": 7, "output_tokens": 3 } + }))) + .mount(&mock) + .await; + + let mut config: AppConfig = toml::from_str( + r#" +[extraction.llm] +api_key = "test-key" +provider = "anthropic" +model = "claude-test" +"#, + ) + .unwrap(); + config.extraction.llm.as_mut().unwrap().base_url = Some(mock.uri()); + let state = AppState::new(config).unwrap(); + let entries = vec![ + PreparedUrl { + url: format!("{}/page-a", mock.uri()), + preflight_error: None, + }, + PreparedUrl { + url: format!("{}/page-b", mock.uri()), + preflight_error: None, + }, + PreparedUrl { + url: format!("{}/page-c", mock.uri()), + preflight_error: None, + }, + ]; + let template = ScrapeRequest { + formats: vec![OutputFormat::Json], + json_schema: Some(json!({ + "type": "object", + "properties": { "title": {"type": "string"} } + })), + ..Default::default() + }; + let id = state.start_extract_job(entries, template).await; + + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let jobs = state.extract_jobs.read().await; + if jobs[&id].claimed_index == Some(1) + && jobs[&id].per_url[0].status == ExtractStatus::Completed + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("second URL was never claimed after the first result persisted"); + let cancelling = state.cancel_extract_job(id).await.unwrap(); + assert_eq!(cancelling.status, ExtractStatus::Cancelling); + + let terminal = tokio::time::timeout(Duration::from_secs(2), async { + loop { + let rec = state.extract_jobs.read().await[&id].clone(); + if rec.status == ExtractStatus::Cancelled { + break rec; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("cancellation barrier never reached terminal"); + assert_eq!(terminal.per_url.len(), 3); + assert_eq!(terminal.per_url[0].status, ExtractStatus::Completed); + assert_eq!(terminal.per_url[0].data, Some(json!({"title": "A"}))); + assert_eq!(terminal.per_url[1].status, ExtractStatus::Completed); + assert_eq!(terminal.per_url[2].status, ExtractStatus::Cancelled); + assert!(terminal.tokens_used > 0); + assert_eq!(terminal.credits_used, 2); + + let requests = mock.received_requests().await.unwrap(); + assert!( + requests + .iter() + .any(|request| request.url.path() == "/page-a") + ); + assert!( + requests + .iter() + .all(|request| request.url.path() != "/page-c"), + "cancellation must prevent the next URL dispatch" + ); +} + #[tokio::test] async fn v1_extract_completes_end_to_end_with_mocked_llm() { use wiremock::matchers::{method, path}; diff --git a/docs/agent-onboarding/SKILL.md b/docs/agent-onboarding/SKILL.md index 8694938e..ebcb2491 100644 --- a/docs/agent-onboarding/SKILL.md +++ b/docs/agent-onboarding/SKILL.md @@ -130,6 +130,17 @@ Parameters: Returns: status and, when complete, a per-URL results array. +### crw_cancel_extract + +Idempotently request cancellation of an extract job. A claimed URL may finish +while status is `cancelling`; terminal `cancelled` preserves that result and +marks every untouched ordered slot `cancelled`. + +Parameters: +- `id` (required) — The extract job ID from `crw_extract` + +Returns the same canonical status envelope as `crw_check_extract_status`. + ### crw_parse_file Parse a local file (PDF) into markdown or structured output without fetching from the web. diff --git a/docs/docs/extract.md b/docs/docs/extract.md index 7db19605..a0513848 100644 --- a/docs/docs/extract.md +++ b/docs/docs/extract.md @@ -215,6 +215,36 @@ If the underlying page scrape is weak, the JSON extraction will also be weak. - Debugging extraction before confirming the underlying scrape succeeded - Using the async `/v1/extract` job for a single URL when a synchronous `/v1/scrape` with `formats: ["json"]` is simpler +## Multi-URL lifecycle and cancellation + +`POST /v1/extract` creates one ordered result slot per requested URL. Poll +`GET /v1/extract/{id}` or cancel idempotently with +`DELETE /v1/extract/{id}`. The lifecycle is: + +```text +processing -> completed | failed +processing -> cancelling -> cancelled +``` + +`cancelling` is not terminal: the one URL already claimed by the sequential +worker may finish and persist its result and measured usage. No later URL is +started. At the terminal barrier, every untouched slot becomes `cancelled` and +omits `data`, `error`, `llmUsage`, and `basis`. The result array always keeps the +original URL count and order. Repeated DELETE returns the persisted state; +DELETE after `completed`, `failed`, or `cancelled` does not rewrite it. + +```bash +curl -X DELETE https://api.fastcrw.com/v1/extract/JOB_ID \ + -H "Authorization: Bearer YOUR_API_KEY" +``` + +The TypeScript SDK exposes `startExtract`, `getExtract`, and `cancelExtract`; +Python exposes `start_extract`, `get_extract`, and `cancel_extract`. Their +convenience `extract` waiter treats `cancelling` as non-terminal, raises a typed +cancellation error with partial results on `cancelled`, and performs a +best-effort DELETE on timeout. Async starters send `Prefer: respond-async` for +managed and self-hosted parity. + ## When to use something else - Use [Scrape](#scraping) when prose or markdown is enough diff --git a/docs/docs/mcp-clients.md b/docs/docs/mcp-clients.md index e72c9187..a303716e 100644 --- a/docs/docs/mcp-clients.md +++ b/docs/docs/mcp-clients.md @@ -26,7 +26,7 @@ Use one of these three patterns: ## What tools you get -Embedded local mode (`crw-mcp`) exposes up to 8 tools: +Embedded local mode (`crw-mcp`) exposes up to 9 tools: - `crw_scrape` - `crw_crawl` @@ -34,10 +34,11 @@ Embedded local mode (`crw-mcp`) exposes up to 8 tools: - `crw_map` - `crw_extract` — structured extraction across URLs (async job) - `crw_check_extract_status` — poll an extract job +- `crw_cancel_extract` — cancel an extract job and return canonical status - `crw_parse_file` — parse a local PDF (base64) to markdown, no OCR (always present) - `crw_search` — only advertised when a search backend is configured; hidden otherwise -fastcrw.com cloud mode exposes all 8 tools: +fastcrw.com cloud mode exposes all 9 tools: - `crw_scrape` - `crw_crawl` @@ -45,6 +46,7 @@ fastcrw.com cloud mode exposes all 8 tools: - `crw_map` - `crw_extract` - `crw_check_extract_status` +- `crw_cancel_extract` - `crw_parse_file` - `crw_search` — always available (managed search backend) @@ -54,7 +56,7 @@ Browser automation mode (`crw-browse`, separate server — v0.4.0+) exposes: - `tree` — accessibility snapshot of the current page :::tip -If you only remember one rule, remember this one: local embedded mode is the easiest setup (up to 8 tools, with `crw_search` appearing automatically when a search backend is configured), and fastcrw.com cloud mode is the easiest way to get all 8 tools including always-on web search. +If you only remember one rule, remember this one: local embedded mode is the easiest setup (up to 9 tools, with `crw_search` appearing automatically when a search backend is configured), and fastcrw.com cloud mode is the easiest way to get all 9 tools including always-on web search. ::: ## Claude Code @@ -230,7 +232,7 @@ Edit `~/.codeium/windsurf/mcp_config.json`. ``` :::note -Windsurf has a total MCP tool limit. CRW stays lightweight: local embedded mode exposes up to 8 tools (`crw_search` is hidden unless a search backend is configured), and cloud mode always exposes all 8 tools. +Windsurf has a total MCP tool limit. CRW stays lightweight: local embedded mode exposes up to 9 tools (`crw_search` is hidden unless a search backend is configured), and cloud mode always exposes all 9 tools. ::: ## Cline @@ -365,7 +367,7 @@ Open Management (`⌘⇧M`) → **Providers** → **MCP Providers** → **Add MC | Command | `npx` | | Args | `crw-mcp` | -`crw_*` tools then appear in chat namespaced as `crw_scrape`, `crw_crawl`, `crw_check_crawl_status`, `crw_map`, `crw_extract`, `crw_check_extract_status`, `crw_parse_file` (and `crw_search` when a search backend is configured). +`crw_*` tools then appear in chat namespaced as `crw_scrape`, `crw_crawl`, `crw_check_crawl_status`, `crw_map`, `crw_extract`, `crw_check_extract_status`, `crw_cancel_extract`, `crw_parse_file` (and `crw_search` when a search backend is configured). ### Or edit the config file diff --git a/docs/docs/mcp.md b/docs/docs/mcp.md index 516a135d..5533ce70 100644 --- a/docs/docs/mcp.md +++ b/docs/docs/mcp.md @@ -10,8 +10,8 @@ CRW includes a built-in MCP (Model Context Protocol) server that gives any MCP-c | Mode | When | Tools | Description | |------|------|-------|-------------| -| **Embedded** (default) | No `--api-url` / `CRW_API_URL` set | scrape, crawl, check_crawl_status, map, extract, check_extract_status, parse_file + **search** (when a search backend is configured) | Self-contained. No server needed. The scraping engine runs inside the MCP process. `crw_search` is advertised only when a search backend is configured (e.g. the Docker compose sidecar). | -| **Proxy / Server** | `--api-url` / `CRW_API_URL` set | scrape, crawl, check_crawl_status, map, extract, check_extract_status, parse_file, search | Forwards tool calls to a remote CRW server — the [fastcrw.com](https://fastcrw.com) cloud **or your own self-hosted server**. `crw_search` is always advertised in proxy mode; it works whenever the server has a search backend configured (the Docker stack enables it by default). | +| **Embedded** (default) | No `--api-url` / `CRW_API_URL` set | scrape, crawl, check_crawl_status, map, extract, check_extract_status, cancel_extract, parse_file + **search** (when a search backend is configured) | Self-contained. No server needed. The scraping engine runs inside the MCP process. `crw_search` is advertised only when a search backend is configured (e.g. the Docker compose sidecar). | +| **Proxy / Server** | `--api-url` / `CRW_API_URL` set | scrape, crawl, check_crawl_status, map, extract, check_extract_status, cancel_extract, parse_file, search | Forwards tool calls to a remote CRW server — the [fastcrw.com](https://fastcrw.com) cloud **or your own self-hosted server**. `crw_search` is always advertised in proxy mode; it works whenever the server has a search backend configured (the Docker stack enables it by default). | ## Where to use what @@ -154,6 +154,7 @@ This yields a ~4.2 MB binary (vs ~17 MB for the default embedded build) because | `crw_map` | Discover all URLs on a site | `POST /v1/map` | All modes | | `crw_extract` | Extract structured JSON from URLs → async job ID | `POST /v1/extract` | All modes | | `crw_check_extract_status` | Poll extract status and get results | `GET /v1/extract/:id` | All modes | +| `crw_cancel_extract` | Idempotently cancel extract and return canonical status | `DELETE /v1/extract/:id` | All modes | | `crw_search` | Search the web → titles, URLs, descriptions | `POST /v1/search` | Always in proxy mode; embedded only when a search backend is configured | | `crw_parse_file` | Parse a local PDF (base64) → markdown | `POST /firecrawl/v2/parse` (multipart) | All modes | diff --git a/docs/openapi-3.0.json b/docs/openapi-3.0.json index 55d30575..25ae5aa6 100644 --- a/docs/openapi-3.0.json +++ b/docs/openapi-3.0.json @@ -406,8 +406,22 @@ "extract" ], "summary": "Start an async multi-URL structured extraction job", - "description": "Native extraction. Returns a per-URL array via GET /v1/extract/{id}. At least one of `prompt` or `schema` is required.", + "description": "Native extraction. Returns a fixed-cardinality per-URL array via GET /v1/extract/{id}. At least one of `prompt` or `schema` is required. SDK async starters send `Prefer: respond-async` for managed/self-hosted parity.", "operationId": "extractStart", + "parameters": [ + { + "name": "Prefer", + "in": "header", + "required": false, + "schema": { + "type": "string", + "enum": [ + "respond-async" + ] + }, + "description": "Request the canonical async lifecycle across managed and self-hosted deployments." + } + ], "requestBody": { "required": true, "content": { @@ -464,6 +478,46 @@ } } }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "404": { + "$ref": "#/components/responses/NotFound" + } + } + }, + "delete": { + "tags": [ + "extract" + ], + "summary": "Cancel an extract job", + "description": "Idempotent. Processing becomes cancelling while a claimed URL settles, then cancelled. Completed, failed, and cancelled states are immutable.", + "operationId": "extractCancel", + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "description": "Extract job id returned by POST /v1/extract" + } + ], + "responses": { + "200": { + "description": "Canonical job status + ordered per-URL results", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ExtractStatus" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, "404": { "$ref": "#/components/responses/NotFound" } @@ -1386,14 +1440,16 @@ "status": { "type": "string", "enum": [ + "processing", "completed", - "failed" + "failed", + "cancelled" ] }, "data": { "type": "object", "additionalProperties": true, - "description": "Extracted JSON object (present when status is completed)." + "description": "Extracted JSON object (present when status is completed). Untouched cancelled slots omit data, error, usage, and basis." }, "error": { "type": "string" @@ -1427,6 +1483,7 @@ "required": [ "id", "status", + "results", "expiresAt", "creditsUsed", "tokensUsed" @@ -1439,8 +1496,10 @@ "type": "string", "enum": [ "processing", + "cancelling", "completed", - "failed" + "failed", + "cancelled" ] }, "results": { @@ -1448,7 +1507,7 @@ "items": { "$ref": "#/components/schemas/ExtractUrlResult" }, - "description": "Per-URL results in original request order." + "description": "Fixed-cardinality per-URL slots in original request order. Pending slots are processing; untouched terminal slots are cancelled." }, "error": { "type": "string", diff --git a/docs/openapi.json b/docs/openapi.json index aebc81aa..2a9ef712 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -777,8 +777,22 @@ "extract" ], "summary": "Start an async multi-URL structured extraction job", - "description": "Native extraction. Returns a per-URL array via GET /v1/extract/{id}. At least one of `prompt` or `schema` is required.", + "description": "Native extraction. Returns a fixed-cardinality per-URL array via GET /v1/extract/{id}. At least one of `prompt` or `schema` is required. SDK async starters send `Prefer: respond-async` for managed/self-hosted parity.", "operationId": "extractStart", + "parameters": [ + { + "name": "Prefer", + "in": "header", + "required": false, + "schema": { + "type": "string", + "enum": [ + "respond-async" + ] + }, + "description": "Request the canonical async lifecycle across managed and self-hosted deployments." + } + ], "requestBody": { "required": true, "content": { @@ -835,6 +849,46 @@ } } }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "404": { + "$ref": "#/components/responses/NotFound" + } + } + }, + "delete": { + "tags": [ + "extract" + ], + "summary": "Cancel an extract job", + "description": "Idempotent. Processing becomes cancelling while a claimed URL settles, then cancelled. Completed, failed, and cancelled states are immutable.", + "operationId": "extractCancel", + "parameters": [ + { + "name": "id", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "description": "Extract job id returned by POST /v1/extract" + } + ], + "responses": { + "200": { + "description": "Canonical job status + ordered per-URL results", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ExtractStatus" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, "404": { "$ref": "#/components/responses/NotFound" } @@ -2006,14 +2060,16 @@ "status": { "type": "string", "enum": [ + "processing", "completed", - "failed" + "failed", + "cancelled" ] }, "data": { "type": "object", "additionalProperties": true, - "description": "Extracted JSON object (present when status is completed)." + "description": "Extracted JSON object (present when status is completed). Untouched cancelled slots omit data, error, usage, and basis." }, "error": { "type": "string" @@ -2047,6 +2103,7 @@ "required": [ "id", "status", + "results", "expiresAt", "creditsUsed", "tokensUsed" @@ -2059,8 +2116,10 @@ "type": "string", "enum": [ "processing", + "cancelling", "completed", - "failed" + "failed", + "cancelled" ] }, "results": { @@ -2068,7 +2127,7 @@ "items": { "$ref": "#/components/schemas/ExtractUrlResult" }, - "description": "Per-URL results in original request order." + "description": "Fixed-cardinality per-URL slots in original request order. Pending slots are processing; untouched terminal slots are cancelled." }, "error": { "type": "string", diff --git a/mcp/crw-mcp/skills/SKILL.md b/mcp/crw-mcp/skills/SKILL.md index 8694938e..ebcb2491 100644 --- a/mcp/crw-mcp/skills/SKILL.md +++ b/mcp/crw-mcp/skills/SKILL.md @@ -130,6 +130,17 @@ Parameters: Returns: status and, when complete, a per-URL results array. +### crw_cancel_extract + +Idempotently request cancellation of an extract job. A claimed URL may finish +while status is `cancelling`; terminal `cancelled` preserves that result and +marks every untouched ordered slot `cancelled`. + +Parameters: +- `id` (required) — The extract job ID from `crw_extract` + +Returns the same canonical status envelope as `crw_check_extract_status`. + ### crw_parse_file Parse a local file (PDF) into markdown or structured output without fetching from the web. diff --git a/sdks/python/README.md b/sdks/python/README.md index 9587df1f..c1ad00d3 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -165,6 +165,15 @@ for r in results: if r["status"] == "completed": print(r["url"], r["data"]) +# Explicit typed lifecycle. start_extract always sends Prefer: respond-async. +accepted = client.start_extract( + ["https://a.example", "https://b.example"], + schema={"type": "object", "properties": {"title": {"type": "string"}}}, + basis=True, +) +status = client.get_extract(accepted["id"]) +client.cancel_extract(accepted["id"]) # idempotent + # Scrape many URLs in one async batch: pages = client.batch_scrape(["https://a.com", "https://b.com"], formats=["markdown"]) diff --git a/sdks/python/src/crw/__init__.py b/sdks/python/src/crw/__init__.py index 35e9338b..748f33a6 100644 --- a/sdks/python/src/crw/__init__.py +++ b/sdks/python/src/crw/__init__.py @@ -1,6 +1,24 @@ """CRW Python SDK — scrape, crawl, and map any website.""" from crw.client import CrwClient -from crw.exceptions import CrwError, CrwApiError, CrwBinaryNotFoundError, CrwTimeoutError +from crw.exceptions import ( + CrwApiError, + CrwBinaryNotFoundError, + CrwError, + CrwExtractCancelledError, + CrwTimeoutError, +) +from crw.types import Basis, ExtractAccepted, ExtractStatus, ExtractUrlResult -__all__ = ["CrwClient", "CrwError", "CrwApiError", "CrwBinaryNotFoundError", "CrwTimeoutError"] +__all__ = [ + "CrwClient", + "CrwError", + "CrwApiError", + "CrwBinaryNotFoundError", + "CrwTimeoutError", + "CrwExtractCancelledError", + "ExtractAccepted", + "ExtractStatus", + "ExtractUrlResult", + "Basis", +] diff --git a/sdks/python/src/crw/client.py b/sdks/python/src/crw/client.py index 886d6020..12a27c3a 100644 --- a/sdks/python/src/crw/client.py +++ b/sdks/python/src/crw/client.py @@ -7,11 +7,12 @@ import os import subprocess import time -from typing import Any +from typing import Any, cast from urllib.parse import quote, urlencode from crw._binary import ensure_binary -from crw.exceptions import CrwApiError, CrwError, CrwTimeoutError +from crw.exceptions import CrwApiError, CrwError, CrwExtractCancelledError, CrwTimeoutError +from crw.types import ExtractAccepted, ExtractStatus, ExtractUrlResult _REQUEST_ID = 0 @@ -447,7 +448,7 @@ def parse_file( args.update(kwargs) return self._tool_call("crw_parse_file", args) - def extract( + def _extract_body( self, urls: list[str], prompt: str | None = None, @@ -455,61 +456,135 @@ def extract( llm_api_key: str | None = None, llm_provider: str | None = None, llm_model: str | None = None, - poll_interval: float = 2.0, - timeout: float = 120.0, - ) -> list[dict]: - """Run native multi-URL structured extraction (HTTP mode only). - - Starts an async ``/v1/extract`` job, polls until completion, and returns - the per-URL ``results`` array (``[{url, status, data, error, llmUsage}]``) - in request order. Requires an LLM configured on the engine (or a BYOK key). - - Args: - urls: URLs to extract from (at least one required). - prompt: Free-text extraction instruction (required unless ``schema``). - schema: JSON Schema describing the desired output shape. - llm_api_key: BYOK LLM key (self-host/local engine only). - llm_provider: BYOK provider. - llm_model: BYOK model. - """ - if not self._api_url: - raise CrwError(_HTTP_ONLY_HINT.format(name="extract", reason="LLM extract job endpoint")) - + basis: bool | None = None, + ) -> dict[str, Any]: body: dict[str, Any] = {"urls": list(urls)} if prompt is not None: body["prompt"] = prompt if schema is not None: body["schema"] = schema + if basis is not None: + body["basis"] = basis if llm_api_key is not None: body["llmApiKey"] = llm_api_key if llm_provider is not None: body["llmProvider"] = llm_provider if llm_model is not None: body["llmModel"] = llm_model + return body + + def _start_extract_request(self, body: dict[str, Any]) -> dict: + return self._http_request( + "POST", + "/v1/extract", + body, + raw=True, + headers={"Prefer": "respond-async"}, + ) + + def start_extract( + self, + urls: list[str], + prompt: str | None = None, + schema: dict | None = None, + llm_api_key: str | None = None, + llm_provider: str | None = None, + llm_model: str | None = None, + basis: bool | None = None, + ) -> ExtractAccepted: + """Start extraction without waiting; always requests an async response.""" + if not self._api_url: + raise CrwError( + _HTTP_ONLY_HINT.format(name="start_extract", reason="LLM extract job endpoint") + ) + result = self._start_extract_request( + self._extract_body( + urls, prompt, schema, llm_api_key, llm_provider, llm_model, basis + ) + ) + if isinstance(result.get("id"), str) and result.get("status") == "processing": + return cast(ExtractAccepted, result) + raise CrwError(f"start_extract did not return an accepted job: {result}") + + def get_extract(self, job_id: str) -> ExtractStatus: + """Return the canonical extract lifecycle envelope.""" + if not self._api_url: + raise CrwError( + _HTTP_ONLY_HINT.format(name="get_extract", reason="LLM extract job endpoint") + ) + return cast( + ExtractStatus, + self._http_request( + "GET", f"/v1/extract/{quote(job_id, safe='')}", raw=True, check_success=False + ), + ) + + def cancel_extract(self, job_id: str) -> ExtractStatus: + """Idempotently request cancellation and return canonical status.""" + if not self._api_url: + raise CrwError( + _HTTP_ONLY_HINT.format(name="cancel_extract", reason="LLM extract job endpoint") + ) + return cast( + ExtractStatus, + self._http_request( + "DELETE", f"/v1/extract/{quote(job_id, safe='')}", raw=True, check_success=False + ), + ) - start = self._http_request("POST", "/v1/extract", body, raw=True) + def extract( + self, + urls: list[str], + prompt: str | None = None, + schema: dict | None = None, + llm_api_key: str | None = None, + llm_provider: str | None = None, + llm_model: str | None = None, + poll_interval: float = 2.0, + timeout: float = 120.0, + basis: bool | None = None, + ) -> list[ExtractUrlResult]: + """Wait for native multi-URL extraction, cancelling best-effort on timeout. + + ``cancelling`` is non-terminal. Terminal ``cancelled`` raises + :class:`CrwExtractCancelledError` with the status and partial results. + """ + if not self._api_url: + raise CrwError(_HTTP_ONLY_HINT.format(name="extract", reason="LLM extract job endpoint")) + + body = self._extract_body( + urls, prompt, schema, llm_api_key, llm_provider, llm_model, basis + ) + start = self._start_extract_request(body) job_id = start.get("id") if not job_id: # The managed API answers synchronously: the extraction is already # finished and the payload is in this first response, with no job to # poll. Only the async path hands back an `id`. Demanding one made # every managed extract() raise. + if start.get("status") == "cancelled": + raise CrwExtractCancelledError(cast(ExtractStatus, start)) if isinstance(start.get("results"), list): - return start["results"] + return cast(list[ExtractUrlResult], start["results"]) raise CrwError(f"extract returned neither a job id nor results: {start}") deadline = time.monotonic() + timeout while True: if time.monotonic() > deadline: + try: + self.cancel_extract(str(job_id)) + except Exception: + # Timeout remains caller-visible; DELETE is best effort. + pass raise CrwTimeoutError(f"Extract {job_id} timed out after {timeout}s") - status_result = self._http_request( - "GET", f"/v1/extract/{job_id}", raw=True, check_success=False - ) + status_result = self.get_extract(str(job_id)) status = status_result.get("status") if status == "completed": - return status_result.get("results", []) + return status_result["results"] if status == "failed": raise CrwError(f"Extract failed: {status_result.get('error', 'unknown')}") + if status == "cancelled": + raise CrwExtractCancelledError(status_result) time.sleep(poll_interval) def batch_scrape( @@ -694,17 +769,18 @@ def _http_request( *, raw: bool = False, check_success: bool = True, + headers: dict[str, str] | None = None, ) -> dict: import urllib.request assert self._api_url is not None url = f"{self._api_url.rstrip('/')}{path}" - headers = {"Content-Type": "application/json"} + request_headers = {"Content-Type": "application/json", **(headers or {})} if self._api_key: - headers["Authorization"] = f"Bearer {self._api_key}" + request_headers["Authorization"] = f"Bearer {self._api_key}" data = json.dumps(body).encode() if body else None - req = urllib.request.Request(url, data=data, headers=headers, method=method) + req = urllib.request.Request(url, data=data, headers=request_headers, method=method) result = _read_json_response(req) diff --git a/sdks/python/src/crw/exceptions.py b/sdks/python/src/crw/exceptions.py index f5d62476..97f07630 100644 --- a/sdks/python/src/crw/exceptions.py +++ b/sdks/python/src/crw/exceptions.py @@ -1,5 +1,10 @@ """CRW SDK exceptions.""" +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from crw.types import ExtractStatus, ExtractUrlResult + class CrwError(Exception): """Base exception for CRW SDK.""" @@ -13,6 +18,15 @@ class CrwTimeoutError(CrwError): """Operation timed out.""" +class CrwExtractCancelledError(CrwError): + """An extract waiter reached the immutable cancelled state.""" + + def __init__(self, status: "ExtractStatus"): + super().__init__(f"Extract {status['id']} was cancelled") + self.status = status + self.results: list["ExtractUrlResult"] = status["results"] + + class CrwApiError(CrwError): """API returned an error.""" diff --git a/sdks/python/src/crw/types.py b/sdks/python/src/crw/types.py new file mode 100644 index 00000000..40f719b9 --- /dev/null +++ b/sdks/python/src/crw/types.py @@ -0,0 +1,68 @@ +"""Typed native extract lifecycle contracts.""" + +from typing import Any, Literal, TypedDict + +ExtractJobState = Literal["processing", "cancelling", "completed", "failed", "cancelled"] +ExtractUrlState = Literal["processing", "completed", "failed", "cancelled"] +FieldStatus = Literal["supported", "unverified", "unsupported", "notFound"] + + +class _EvidenceCitationRequired(TypedDict): + url: str + sourceHash: str + sourceTextKind: str + + +class EvidenceCitation(_EvidenceCitationRequired, total=False): + title: str + excerpt: str + + +class _BasisRequired(TypedDict): + basisVersion: int + field: str + value: Any | None + status: FieldStatus + citations: list[EvidenceCitation] + + +class Basis(_BasisRequired, total=False): + confidence: Literal["low", "medium", "high"] + + +class BasisWarning(TypedDict): + field: str + code: str + + +class ExtractAccepted(TypedDict): + id: str + status: Literal["processing"] + urls: int + + +class _ExtractUrlResultRequired(TypedDict): + url: str + status: ExtractUrlState + + +class ExtractUrlResult(_ExtractUrlResultRequired, total=False): + data: Any + error: str + llmUsage: dict[str, Any] + basis: list[Basis] + basisWarnings: list[BasisWarning] + llmInputHash: str + + +class _ExtractStatusRequired(TypedDict): + id: str + status: ExtractJobState + results: list[ExtractUrlResult] + expiresAt: str + creditsUsed: int + tokensUsed: int + + +class ExtractStatus(_ExtractStatusRequired, total=False): + error: str diff --git a/sdks/python/tests/test_client_unit.py b/sdks/python/tests/test_client_unit.py index ff9a298c..3bf6d6b7 100644 --- a/sdks/python/tests/test_client_unit.py +++ b/sdks/python/tests/test_client_unit.py @@ -8,7 +8,7 @@ import pytest from crw.client import CLOUD_API_URL, CrwClient -from crw.exceptions import CrwApiError, CrwError, CrwTimeoutError +from crw.exceptions import CrwApiError, CrwError, CrwExtractCancelledError, CrwTimeoutError @pytest.fixture(autouse=True) @@ -195,16 +195,20 @@ def test_search_http_sends_query(self) -> None: client = CrwClient(api_url="https://fastcrw.com/api", api_key="fc-test") mock_response = [{"url": "https://example.com", "title": "Example"}] - with patch.object(client, "_http_post", return_value=mock_response) as mock_post: + with patch.object( + client, + "_http_request", + return_value={"success": True, "data": mock_response}, + ) as mock_request: result = client.search("python web scraping", limit=5) - mock_post.assert_called_once() - call_args = mock_post.call_args - assert call_args[0][0] == "/v1/search" - body = call_args[0][1] - assert body["query"] == "python web scraping" - assert body["limit"] == 5 - assert result == mock_response + mock_request.assert_called_once_with( + "POST", + "/v1/search", + {"query": "python web scraping", "limit": 5}, + raw=True, + ) + assert list(result) == mock_response # --------------------------------------------------------------------------- @@ -342,6 +346,9 @@ class TestHttpOnlyMethods: "call", [ lambda c: c.extract(["https://example.com"], prompt="x"), + lambda c: c.start_extract(["https://example.com"], prompt="x"), + lambda c: c.get_extract("job-1"), + lambda c: c.cancel_extract("job-1"), lambda c: c.batch_scrape(["https://example.com"]), lambda c: c.capabilities(), lambda c: c.change_tracking_diff({"markdown": "a"}), @@ -373,6 +380,100 @@ def test_extract_polls_until_complete(self) -> None: ] # Native route, not the FC-legacy /v2/extract. assert req.call_args_list[0][0][:2] == ("POST", "/v1/extract") + assert req.call_args_list[0].kwargs["headers"] == {"Prefer": "respond-async"} + + def test_start_extract_prefer_managed_and_self_hosted_fixtures(self) -> None: + accepted = {"id": "job-1", "status": "processing", "urls": 1} + for client in ( + CrwClient(api_key="fc-test"), + CrwClient(api_url="http://localhost:3000"), + ): + with patch.object(client, "_http_request", return_value=accepted) as req: + result = client.start_extract( + ["https://example.com"], schema={"type": "object"}, basis=True + ) + assert result["id"] == "job-1" + assert req.call_args[0][:2] == ("POST", "/v1/extract") + assert req.call_args.kwargs["headers"] == {"Prefer": "respond-async"} + assert req.call_args[0][2]["basis"] is True + + def test_extract_preserves_managed_sync_fixture_with_prefer(self) -> None: + client = CrwClient(api_key="fc-test") + sync = { + "success": True, + "results": [ + {"url": "https://example.com", "status": "completed", "data": {"title": "Hi"}} + ], + } + with patch.object(client, "_http_request", return_value=sync) as req: + result = client.extract(["https://example.com"], prompt="title") + assert result[0]["status"] == "completed" + assert req.call_args.kwargs["headers"] == {"Prefer": "respond-async"} + + def test_get_and_cancel_extract_use_canonical_route(self) -> None: + status = { + "id": "job/one", + "status": "cancelled", + "results": [{"url": "https://example.com", "status": "cancelled"}], + "expiresAt": "2026-07-14T00:00:00Z", + "creditsUsed": 0, + "tokensUsed": 0, + } + client = CrwClient(api_url="http://localhost:3000") + with patch.object(client, "_http_request", return_value=status) as req: + assert client.get_extract("job/one")["status"] == "cancelled" + assert client.cancel_extract("job/one")["results"][0]["status"] == "cancelled" + assert req.call_args_list[0][0][:2] == ("GET", "/v1/extract/job%2Fone") + assert req.call_args_list[1][0][:2] == ("DELETE", "/v1/extract/job%2Fone") + + def test_extract_cancelling_then_typed_cancelled_error(self) -> None: + client = CrwClient(api_url="http://localhost:3000") + accepted = {"id": "job-1", "status": "processing", "urls": 2} + cancelling = { + "id": "job-1", + "status": "cancelling", + "results": [ + {"url": "https://a.example", "status": "completed", "data": {"title": "A"}}, + {"url": "https://b.example", "status": "processing"}, + ], + "expiresAt": "2026-07-14T00:00:00Z", + "creditsUsed": 1, + "tokensUsed": 9, + } + cancelled = { + **cancelling, + "status": "cancelled", + "results": [ + {"url": "https://a.example", "status": "completed", "data": {"title": "A"}}, + {"url": "https://b.example", "status": "cancelled"}, + ], + } + with patch.object(client, "_http_request", side_effect=[accepted, cancelling, cancelled]): + with patch("time.sleep"): + with pytest.raises(CrwExtractCancelledError) as exc: + client.extract( + ["https://a.example", "https://b.example"], + prompt="x", + poll_interval=0, + ) + assert exc.value.status["status"] == "cancelled" + assert exc.value.results[0]["status"] == "completed" + + def test_extract_timeout_best_effort_deletes(self) -> None: + client = CrwClient(api_url="http://localhost:3000") + accepted = {"id": "job-1", "status": "processing", "urls": 1} + cancelled = { + "id": "job-1", + "status": "cancelled", + "results": [{"url": "https://example.com", "status": "cancelled"}], + "expiresAt": "2026-07-14T00:00:00Z", + "creditsUsed": 0, + "tokensUsed": 0, + } + with patch.object(client, "_http_request", side_effect=[accepted, cancelled]) as req: + with pytest.raises(CrwTimeoutError): + client.extract(["https://example.com"], prompt="x", timeout=-1) + assert req.call_args_list[-1][0][:2] == ("DELETE", "/v1/extract/job-1") def test_batch_scrape_returns_pages(self) -> None: client = CrwClient(api_url="https://fastcrw.com/api") diff --git a/sdks/typescript/README.md b/sdks/typescript/README.md index f4871b13..96e7b58a 100644 --- a/sdks/typescript/README.md +++ b/sdks/typescript/README.md @@ -53,6 +53,7 @@ CRW_LOCAL=1 node app.js | `search(query, opts?)` | Web search (+ optional scrape) | both¹ | | `parseFile(bytes, opts?)` | PDF → markdown / structured JSON | both | | `extract({urls, schema?})` | Structured LLM extraction | HTTP | +| `startExtract(opts)` / `getExtract(id)` / `cancelExtract(id)` | Typed extract lifecycle | HTTP | | `batchScrape(urls, opts?)` | Scrape many URLs (async) | HTTP | | `capabilities()` | Feature-detect the engine | HTTP | | `changeTrackingDiff(cur, prev?)` | Diff vs a prior snapshot | HTTP | @@ -69,6 +70,14 @@ const results = await crw.extract({ // results: [{ url, status, data, error, llmUsage }] for (const r of results) if (r.status === "completed") console.log(r.url, r.data); +// Explicit lifecycle. startExtract always sends Prefer: respond-async. +const accepted = await crw.startExtract({ + urls: ["https://a.example", "https://b.example"], + schema: { type: "object", properties: { title: { type: "string" } } }, +}); +const status = await crw.getExtract(accepted.id); +await crw.cancelExtract(accepted.id); // idempotent + // Parse a PDF: import { readFileSync } from "node:fs"; const doc = await crw.parseFile(readFileSync("invoice.pdf"), { formats: ["markdown"] }); diff --git a/sdks/typescript/src/client.ts b/sdks/typescript/src/client.ts index de8b2548..e91af1d2 100644 --- a/sdks/typescript/src/client.ts +++ b/sdks/typescript/src/client.ts @@ -1,6 +1,6 @@ /** CRW client — cloud (default), self-hosted HTTP, or local subprocess mode. */ -import { CrwApiError, CrwError, CrwTimeoutError } from "./errors.js"; +import { CrwApiError, CrwError, CrwExtractCancelledError, CrwTimeoutError } from "./errors.js"; import { LocalTransport } from "./local.js"; import type { BatchResult, @@ -12,7 +12,9 @@ import type { CrawlResult, DiffResult, ExtractOptions, + ExtractAccepted, ExtractResult, + ExtractStatus, Json, MapOptions, ParseFileOptions, @@ -221,39 +223,91 @@ export class CrwClient { } /** - * Native multi-URL structured extraction (HTTP mode only). Starts an async - * `/v1/extract` job and polls until complete, returning a per-URL results - * array (`[{ url, status, data, error, llmUsage }]`) in request order. + * Start native multi-URL extraction without waiting. `Prefer: respond-async` + * is always sent so managed and self-hosted servers use the same lifecycle. + */ + async startExtract(opts: ExtractOptions): Promise { + if (!this.apiUrl) throw new CrwError(httpOnlyHint("startExtract", "LLM extract job endpoint")); + const start = await this.startExtractRequest(opts); + if (typeof start.id === "string" && start.status === "processing") { + return start as unknown as ExtractAccepted; + } + throw new CrwError(`startExtract did not return an accepted job: ${JSON.stringify(start)}`); + } + + /** Get the canonical lifecycle envelope for an extract job. */ + async getExtract(id: string): Promise { + if (!this.apiUrl) throw new CrwError(httpOnlyHint("getExtract", "LLM extract job endpoint")); + const result = await this.httpRequest("GET", `/v1/extract/${encodeURIComponent(id)}`, undefined, { + raw: true, + checkSuccess: false, + }); + return result as unknown as ExtractStatus; + } + + /** Idempotently request cancellation and return the canonical lifecycle envelope. */ + async cancelExtract(id: string): Promise { + if (!this.apiUrl) throw new CrwError(httpOnlyHint("cancelExtract", "LLM extract job endpoint")); + const result = await this.httpRequest("DELETE", `/v1/extract/${encodeURIComponent(id)}`, undefined, { + raw: true, + checkSuccess: false, + }); + return result as unknown as ExtractStatus; + } + + /** + * Convenience waiter over start/get/cancel. `cancelling` is non-terminal; + * terminal cancellation raises a typed error carrying partial results. */ async extract(opts: ExtractOptions): Promise { if (!this.apiUrl) throw new CrwError(httpOnlyHint("extract", "LLM extract job endpoint")); - const { urls, prompt, schema, basis, llmApiKey, llmProvider, llmModel, pollInterval = 2, timeout = 120 } = opts; - const body: Json = { urls: [...urls] }; - if (prompt !== undefined) body.prompt = prompt; - if (schema !== undefined) body.schema = schema; - if (basis !== undefined) body.basis = basis; - if (llmApiKey !== undefined) body.llmApiKey = llmApiKey; - if (llmProvider !== undefined) body.llmProvider = llmProvider; - if (llmModel !== undefined) body.llmModel = llmModel; - const start = await this.httpRequest("POST", "/v1/extract", body, { raw: true }); + const { pollInterval = 2, timeout = 120 } = opts; + const start = await this.startExtractRequest(opts); const jobId = start.id as string | undefined; - // The managed API answers synchronously: the extraction is already finished and - // the payload is in this first response, with no job to poll. Only the async - // path hands back an `id`. Demanding one made every managed extract() throw. + // Preserve the managed synchronous response during rollout if an older + // endpoint ignores Prefer. New managed/self-hosted fixtures return an id. if (!jobId) { - if (Array.isArray(start.results)) return start.results as Json[]; + if (start.status === "cancelled") { + throw new CrwExtractCancelledError(start as unknown as ExtractStatus); + } + if (Array.isArray(start.results)) return start.results as ExtractResult; throw new CrwError(`extract returned neither a job id nor results: ${JSON.stringify(start)}`); } const deadline = Date.now() + timeout * 1000; for (;;) { - if (Date.now() > deadline) throw new CrwTimeoutError(`Extract ${jobId} timed out after ${timeout}s`); - const status = await this.httpRequest("GET", `/v1/extract/${jobId}`, undefined, { raw: true, checkSuccess: false }); - if (status.status === "completed") return (status.results as Json[]) ?? []; + if (Date.now() > deadline) { + try { + await this.cancelExtract(jobId); + } catch { + // Timeout remains the caller-visible failure; DELETE is best effort. + } + throw new CrwTimeoutError(`Extract ${jobId} timed out after ${timeout}s`); + } + const status = await this.getExtract(jobId); + if (status.status === "completed") return status.results; if (status.status === "failed") throw new CrwError(`Extract failed: ${status.error ?? "unknown"}`); + if (status.status === "cancelled") throw new CrwExtractCancelledError(status); await sleep(pollInterval * 1000); } } + private async startExtractRequest(opts: ExtractOptions): Promise { + const { urls, prompt, schema, basis, llmApiKey, llmProvider, llmModel, pollInterval = 2, timeout = 120 } = opts; + void pollInterval; + void timeout; + const body: Json = { urls: [...urls] }; + if (prompt !== undefined) body.prompt = prompt; + if (schema !== undefined) body.schema = schema; + if (basis !== undefined) body.basis = basis; + if (llmApiKey !== undefined) body.llmApiKey = llmApiKey; + if (llmProvider !== undefined) body.llmProvider = llmProvider; + if (llmModel !== undefined) body.llmModel = llmModel; + return this.httpRequest("POST", "/v1/extract", body, { + raw: true, + headers: { Prefer: "respond-async" }, + }); + } + /** Scrape many URLs in one async batch job (HTTP mode only). */ async batchScrape(urls: string[], opts: BatchScrapeOptions = {}): Promise { if (!this.apiUrl) throw new CrwError(httpOnlyHint("batchScrape", "batch job endpoint")); @@ -321,11 +375,18 @@ export class CrwClient { method: string, path: string, body?: Json, - { raw = false, checkSuccess = true }: { raw?: boolean; checkSuccess?: boolean } = {}, + { + raw = false, + checkSuccess = true, + headers: extraHeaders, + }: { raw?: boolean; checkSuccess?: boolean; headers?: Record } = {}, ): Promise { if (this.apiUrl === null) throw new CrwError("internal: httpRequest in local mode"); const url = `${this.apiUrl.replace(/\/$/, "")}${path}`; - const headers: Record = { "Content-Type": "application/json" }; + const headers: Record = { + "Content-Type": "application/json", + ...extraHeaders, + }; if (this.apiKey) headers.Authorization = `Bearer ${this.apiKey}`; const resp = await fetch(url, { method, headers, body: body ? JSON.stringify(body) : undefined }); const result = await this.readJson(resp); diff --git a/sdks/typescript/src/errors.ts b/sdks/typescript/src/errors.ts index f9e5ef07..b4069d06 100644 --- a/sdks/typescript/src/errors.ts +++ b/sdks/typescript/src/errors.ts @@ -1,5 +1,7 @@ /** CRW SDK error types. */ +import type { ExtractStatus } from "./types.js"; + export class CrwError extends Error { constructor(message: string) { super(message); @@ -23,6 +25,19 @@ export class CrwTimeoutError extends CrwError { } } +/** A convenience extract waiter reached the immutable cancelled state. */ +export class CrwExtractCancelledError extends CrwError { + readonly status: ExtractStatus; + readonly results: ExtractStatus["results"]; + + constructor(status: ExtractStatus) { + super(`Extract ${status.id} was cancelled`); + this.name = "CrwExtractCancelledError"; + this.status = status; + this.results = status.results; + } +} + export class CrwBinaryNotFoundError extends CrwError { constructor(message: string) { super(message); diff --git a/sdks/typescript/src/index.ts b/sdks/typescript/src/index.ts index da789cd2..80eb0443 100644 --- a/sdks/typescript/src/index.ts +++ b/sdks/typescript/src/index.ts @@ -1,5 +1,11 @@ export { CrwClient, CLOUD_API_URL, DASHBOARD_URL, DOCS_URL } from "./client.js"; -export { CrwError, CrwApiError, CrwTimeoutError, CrwBinaryNotFoundError } from "./errors.js"; +export { + CrwError, + CrwApiError, + CrwTimeoutError, + CrwExtractCancelledError, + CrwBinaryNotFoundError, +} from "./errors.js"; export type { ClientOptions, ScrapeOptions, @@ -8,6 +14,15 @@ export type { SearchOptions, ParseFileOptions, ExtractOptions, + ExtractJobState, + ExtractUrlState, + ExtractAccepted, + ExtractStatus, + ExtractUrlResult, + Basis, + BasisWarning, + EvidenceCitation, + FieldStatus, BatchScrapeOptions, ChangeTrackingOptions, ScrapeResult, diff --git a/sdks/typescript/src/types.ts b/sdks/typescript/src/types.ts index 76d656ac..19b2eb84 100644 --- a/sdks/typescript/src/types.ts +++ b/sdks/typescript/src/types.ts @@ -129,6 +129,46 @@ export interface ExtractOptions { timeout?: number; } +export type ExtractJobState = + | "processing" + | "cancelling" + | "completed" + | "failed" + | "cancelled"; + +export type ExtractUrlState = "processing" | "completed" | "failed" | "cancelled"; + +/** Response from an async extract admission request. */ +export interface ExtractAccepted { + id: string; + status: "processing"; + /** Count of URLs enqueued for fetch. */ + urls: number; +} + +/** One fixed, ordered result slot for a requested URL. */ +export interface ExtractUrlResult { + url: string; + status: ExtractUrlState; + data?: unknown; + error?: string; + llmUsage?: Json; + basis?: Basis[]; + basisWarnings?: BasisWarning[]; + llmInputHash?: string; +} + +/** Canonical GET/DELETE extract lifecycle envelope. */ +export interface ExtractStatus { + id: string; + status: ExtractJobState; + results: ExtractUrlResult[]; + error?: string; + expiresAt: string; + creditsUsed: number; + tokensUsed: number; +} + export interface BatchScrapeOptions { formats?: string[]; pollInterval?: number; @@ -148,7 +188,7 @@ export type CrawlResult = Json[]; export type SearchResult = Json | Json[]; export type ParseResult = Json; /** Native `/v1/extract` returns one result object per URL, in request order. */ -export type ExtractResult = Json[]; +export type ExtractResult = ExtractUrlResult[]; export type BatchResult = Json[]; export type Capabilities = Json; export type DiffResult = Json; diff --git a/sdks/typescript/test/client.test.ts b/sdks/typescript/test/client.test.ts index e9ea10e9..77a6c87e 100644 --- a/sdks/typescript/test/client.test.ts +++ b/sdks/typescript/test/client.test.ts @@ -1,6 +1,12 @@ import assert from "node:assert/strict"; import { afterEach, beforeEach, test } from "node:test"; -import { CLOUD_API_URL, CrwClient, CrwError } from "../dist/esm/index.js"; +import { + CLOUD_API_URL, + CrwClient, + CrwError, + CrwExtractCancelledError, + CrwTimeoutError, +} from "../dist/esm/index.js"; const origFetch = globalThis.fetch; const origEnv = { ...process.env }; @@ -66,6 +72,9 @@ test("HTTP-only methods throw in local mode", async () => { process.env.CRW_LOCAL = "1"; const c = new CrwClient(); await assert.rejects(() => c.extract({ urls: ["https://example.com"] }), /requires HTTP mode/); + await assert.rejects(() => c.startExtract({ urls: ["https://example.com"] }), /requires HTTP mode/); + await assert.rejects(() => c.getExtract("job-1"), /requires HTTP mode/); + await assert.rejects(() => c.cancelExtract("job-1"), /requires HTTP mode/); await assert.rejects(() => c.batchScrape(["https://example.com"]), /requires HTTP mode/); await assert.rejects(() => c.capabilities(), /requires HTTP mode/); await assert.rejects(() => c.changeTrackingDiff({ markdown: "a" }), /requires HTTP mode/); @@ -100,6 +109,7 @@ test("extract starts a /v1/extract job and returns per-URL results", async () => const postBody = JSON.parse(calls[0].init!.body as string); assert.deepEqual(postBody.urls, ["https://example.com"]); assert.equal(postBody.llmApiKey, "sk"); + assert.equal((calls[0].init!.headers as Record).Prefer, "respond-async"); assert.ok(calls[1].url.startsWith(`${CLOUD_API_URL}/v1/extract/job-1`)); assert.equal(results.length, 1); const r0 = results[0] as { url: string; data: { title: string } }; @@ -107,6 +117,115 @@ test("extract starts a /v1/extract job and returns per-URL results", async () => assert.equal(r0.data.title, "Hi"); }); +test("startExtract sends Prefer for managed and self-hosted fixtures", async () => { + const calls = mockFetch({ id: "job-1", status: "processing", urls: 1 }); + const managed = new CrwClient({ apiKey: "fc-test" }); + assert.equal((await managed.startExtract({ urls: ["https://example.com"], prompt: "x" })).id, "job-1"); + const selfHosted = new CrwClient({ apiUrl: "http://localhost:3000" }); + await selfHosted.startExtract({ urls: ["https://example.com"], schema: { type: "object" } }); + + assert.equal(calls[0].url, `${CLOUD_API_URL}/v1/extract`); + assert.equal(calls[1].url, "http://localhost:3000/v1/extract"); + for (const call of calls) { + assert.equal((call.init!.headers as Record).Prefer, "respond-async"); + } +}); + +test("extract preserves managed synchronous fixture while requesting async", async () => { + const calls = mockFetch({ + success: true, + results: [{ url: "https://example.com", status: "completed", data: { title: "Hi" } }], + }); + const c = new CrwClient({ apiKey: "fc-test" }); + const results = await c.extract({ urls: ["https://example.com"], prompt: "title" }); + assert.equal(results[0].status, "completed"); + assert.equal((calls[0].init!.headers as Record).Prefer, "respond-async"); +}); + +test("getExtract and cancelExtract use canonical status route", async () => { + const status = { + id: "job/one", + status: "cancelled", + results: [{ url: "https://example.com", status: "cancelled" }], + expiresAt: "2026-07-14T00:00:00Z", + creditsUsed: 0, + tokensUsed: 0, + }; + const calls = mockFetch(status); + const c = new CrwClient({ apiUrl: "http://localhost:3000" }); + assert.equal((await c.getExtract("job/one")).status, "cancelled"); + assert.equal((await c.cancelExtract("job/one")).results.length, 1); + assert.equal(calls[0].url, "http://localhost:3000/v1/extract/job%2Fone"); + assert.equal(calls[0].init!.method, "GET"); + assert.equal(calls[1].init!.method, "DELETE"); +}); + +test("extract treats cancelling as non-terminal then raises typed cancelled error", async () => { + const responses = [ + { id: "job-1", status: "processing", urls: 2 }, + { + id: "job-1", + status: "cancelling", + results: [ + { url: "https://a.example", status: "completed", data: { title: "A" } }, + { url: "https://b.example", status: "processing" }, + ], + expiresAt: "2026-07-14T00:00:00Z", + creditsUsed: 1, + tokensUsed: 9, + }, + { + id: "job-1", + status: "cancelled", + results: [ + { url: "https://a.example", status: "completed", data: { title: "A" } }, + { url: "https://b.example", status: "cancelled" }, + ], + expiresAt: "2026-07-14T00:00:00Z", + creditsUsed: 1, + tokensUsed: 9, + }, + ]; + globalThis.fetch = (async () => { + const body = responses.shift(); + return { ok: true, status: 200, statusText: "OK", text: async () => JSON.stringify(body) } as Response; + }) as typeof fetch; + const c = new CrwClient({ apiUrl: "http://localhost:3000" }); + await assert.rejects( + () => c.extract({ urls: ["https://a.example", "https://b.example"], prompt: "x", pollInterval: 0 }), + (error: unknown) => { + assert.ok(error instanceof CrwExtractCancelledError); + assert.equal(error.status.status, "cancelled"); + assert.equal(error.results[0].status, "completed"); + return true; + }, + ); +}); + +test("extract timeout performs best-effort DELETE", async () => { + const calls: Array<{ url: string; init?: RequestInit }> = []; + globalThis.fetch = (async (url: string, init?: RequestInit) => { + calls.push({ url: String(url), init }); + const body = init?.method === "POST" + ? { id: "job-1", status: "processing", urls: 1 } + : { + id: "job-1", + status: "cancelled", + results: [{ url: "https://example.com", status: "cancelled" }], + expiresAt: "2026-07-14T00:00:00Z", + creditsUsed: 0, + tokensUsed: 0, + }; + return { ok: true, status: 200, statusText: "OK", text: async () => JSON.stringify(body) } as Response; + }) as typeof fetch; + const c = new CrwClient({ apiUrl: "http://localhost:3000" }); + await assert.rejects( + () => c.extract({ urls: ["https://example.com"], prompt: "x", timeout: -1 }), + CrwTimeoutError, + ); + assert.equal(calls.at(-1)?.init?.method, "DELETE"); +}); + test("capabilities unwraps and uses GET /v1/capabilities", async () => { const calls = mockFetch({ version: "0.14.0" }); const c = new CrwClient({ apiKey: "fc-test" });