diff --git a/README.md b/README.md index 38f7a07..51c2def 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,48 @@ cargo run -p hydra-codegen -- check # CI guard: fails if artifacts are stale 4. `include!` the generated files, implement one dispatch function, and wire your binaries. See `examples/notes/src/lib.rs`. +### JSON batch operations + +Hydra v0.2.0 can project a declared `json` body parameter to HTTP, MCP, and +CLI. Declare the JSON Schema explicitly; Hydra embeds it in the MCP input +schema and generates the HTTP route from the same operation. For a batch that +needs a shell-friendly CLI representation, declare the representation rather +than inferring one: + +```yaml +- name: ingest_batch + description: Apply an ordered, replayable source batch. + method: POST + path: /ingest/batches + read: false + output_type: IngestReceipt + parameters: + - name: replay_key + description: Stable idempotency key. + type: string + required: true + location: body + - name: events + description: Ordered source events. + type: json + required: true + location: body + schema: + type: array + minItems: 1 + items: { type: object } + cli: + flag: event + multiple: true +``` + +HTTP and MCP callers pass `events` as the declared JSON array. The generated +CLI accepts repeated `--event ''` flags; the consumer's single +dispatch function parses that explicit CLI representation before typed +validation and persistence. Hydra does not own source-specific event models, +batch hashing, idempotency, or transactions. `examples/notes` contains a +tested `ingest_batch` reference operation and is the pattern Iris should use. + ## Raw-request (webhook) operations Operations that must see the exact wire representation — signature-verified diff --git a/examples/notes/api/operations.yaml b/examples/notes/api/operations.yaml index a29ecf7..b0d2bc4 100644 --- a/examples/notes/api/operations.yaml +++ b/examples/notes/api/operations.yaml @@ -126,3 +126,33 @@ operations: - flag: attach-mime field: attach_mime description: MIME type for the corresponding local-path --attach value. + - name: ingest_batch + description: Accept a source-agnostic, replayable batch of JSON events through every generated surface. + method: POST + path: /ingest/batches + read: false + output_type: IngestReceipt + parameters: + - name: replay_key + description: Stable source replay key used by the consumer for idempotency. + type: string + required: true + location: body + - name: batch_hash + description: Content hash for replay-conflict detection, computed by the consumer. + type: string + required: true + location: body + - name: events + description: Ordered source events in the batch. + type: json + required: true + location: body + schema: + type: array + minItems: 1 + items: + type: object + cli: + flag: event + multiple: true diff --git a/examples/notes/generated/cli.rs b/examples/notes/generated/cli.rs index 1209d7c..8df5be6 100644 --- a/examples/notes/generated/cli.rs +++ b/examples/notes/generated/cli.rs @@ -18,6 +18,8 @@ pub enum GeneratedCommand { CompactNotes(CompactNotesArgs), /// Append an annotation to a note, optionally carrying attachments. AnnotateNote(AnnotateNoteArgs), + /// Accept a source-agnostic, replayable batch of JSON events through every generated surface. + IngestBatch(IngestBatchArgs), } impl GeneratedCommand { @@ -29,6 +31,7 @@ impl GeneratedCommand { Self::DeleteNote(_) => "delete_note", Self::CompactNotes(_) => "compact_notes", Self::AnnotateNote(_) => "annotate_note", + Self::IngestBatch(_) => "ingest_batch", } } @@ -40,6 +43,7 @@ impl GeneratedCommand { Self::DeleteNote(args) => serde_json::json!({"note_id": args.note_id.clone()}), Self::CompactNotes(_args) => serde_json::json!({}), Self::AnnotateNote(args) => serde_json::json!({"note_id": args.note_id.clone(), "body": args.body.clone(), "attachments": args.attachments.clone().unwrap_or_default(), "attach_mime": args.attach_mime.clone()}), + Self::IngestBatch(args) => serde_json::json!({"replay_key": args.replay_key.clone(), "batch_hash": args.batch_hash.clone(), "events": args.events.clone().unwrap_or_default()}), } } } @@ -92,3 +96,16 @@ pub struct AnnotateNoteArgs { pub attach_mime: Option>, } +#[derive(Debug, Clone, Serialize, Deserialize, Args)] +pub struct IngestBatchArgs { + /// Stable source replay key used by the consumer for idempotency. + #[arg(long)] + pub replay_key: String, + /// Content hash for replay-conflict detection, computed by the consumer. + #[arg(long)] + pub batch_hash: String, + /// Ordered source events in the batch. + #[arg(long = "event", action = clap::ArgAction::Append, required = true)] + pub events: Option>, +} + diff --git a/examples/notes/generated/http.rs b/examples/notes/generated/http.rs index 07e44af..dd5cd3e 100644 --- a/examples/notes/generated/http.rs +++ b/examples/notes/generated/http.rs @@ -46,6 +46,7 @@ pub const GENERATED_ROUTES: &[GeneratedRoute] = &[ GeneratedRoute { name: "note_stats", method: "GET", path: "/stats" }, GeneratedRoute { name: "echo_raw", method: "POST", path: "/hooks/echo" }, GeneratedRoute { name: "annotate_note", method: "POST", path: "/notes/{note_id}/annotate" }, + GeneratedRoute { name: "ingest_batch", method: "POST", path: "/ingest/batches" }, ]; pub fn generated_router() -> Router { @@ -57,6 +58,7 @@ pub fn generated_router() -> Router { .route("/stats", get(note_stats)) .route("/hooks/echo", post(echo_raw)) .route("/notes/{note_id}/annotate", post(annotate_note)) + .route("/ingest/batches", post(ingest_batch)) } async fn list_notes( @@ -181,3 +183,19 @@ async fn annotate_note( ) .await } + +async fn ingest_batch( + State(state): State, + Json(body): Json, +) -> Response { + crate::execute_operation_http( + &state, + "ingest_batch", + GeneratedOperationInput { + path: BTreeMap::new(), + query: BTreeMap::new(), + body, + }, + ) + .await +} diff --git a/examples/notes/generated/mcp.json b/examples/notes/generated/mcp.json index 4b62eb4..7af65da 100644 --- a/examples/notes/generated/mcp.json +++ b/examples/notes/generated/mcp.json @@ -15,6 +15,11 @@ "get_note": { "note_id": "path" }, + "ingest_batch": { + "batch_hash": "body", + "events": "body", + "replay_key": "body" + }, "list_notes": { "limit": "query" }, @@ -162,6 +167,37 @@ "type": "object" }, "name": "annotate_note" + }, + { + "description": "Accept a source-agnostic, replayable batch of JSON events through every generated surface.", + "inputSchema": { + "additionalProperties": false, + "properties": { + "batch_hash": { + "description": "Content hash for replay-conflict detection, computed by the consumer.", + "type": "string" + }, + "events": { + "description": "Ordered source events in the batch.", + "items": { + "type": "object" + }, + "minItems": 1, + "type": "array" + }, + "replay_key": { + "description": "Stable source replay key used by the consumer for idempotency.", + "type": "string" + } + }, + "required": [ + "replay_key", + "batch_hash", + "events" + ], + "type": "object" + }, + "name": "ingest_batch" } ] } diff --git a/examples/notes/src/lib.rs b/examples/notes/src/lib.rs index ddf571b..0e5fd95 100644 --- a/examples/notes/src/lib.rs +++ b/examples/notes/src/lib.rs @@ -186,6 +186,7 @@ pub async fn execute_operation( "attachments": attachments, })) } + "ingest_batch" => execute_ingest_batch(input.body), "note_stats" => { let out_stats = { let notes = state.notes.lock().expect("notes lock"); @@ -285,6 +286,53 @@ fn summarize_attachments(attachments: &Value) -> Result, OperationErr Ok(summaries) } +/// Apply the reference batch contract after every generated surface reaches +/// this single dispatch function. Persistence and idempotency stay consumer-owned. +fn execute_ingest_batch(body: Value) -> Result { + #[derive(Deserialize)] + struct IngestArgs { + replay_key: String, + batch_hash: String, + events: Value, + } + let args: IngestArgs = serde_json::from_value(body) + .map_err(|e| bad_request(&format!("invalid batch body: {e}")))?; + if args.replay_key.trim().is_empty() || args.batch_hash.trim().is_empty() { + return Err(bad_request("replay_key and batch_hash must not be blank")); + } + let events = normalize_batch_events(&args.events)?; + Ok(serde_json::json!({ + "accepted": events.len(), + "replay_key": args.replay_key, + "batch_hash": args.batch_hash, + })) +} + +/// Normalize repeated CLI `--event ` values to the ordered JSON objects +/// HTTP and MCP callers pass directly in the declared batch field. +fn normalize_batch_events(events: &Value) -> Result, OperationError> { + let Some(events) = events.as_array() else { + return Err(bad_request("events must be a non-empty array")); + }; + if events.is_empty() { + return Err(bad_request("events must be a non-empty array")); + } + + let mut normalized = Vec::with_capacity(events.len()); + for event in events { + let event = match event { + Value::String(raw) => serde_json::from_str(raw) + .map_err(|e| bad_request(&format!("--event must be JSON: {e}")))?, + event => event.clone(), + }; + if !event.is_object() { + return Err(bad_request("each event must be a JSON object")); + } + normalized.push(event); + } + Ok(normalized) +} + fn bad_request(message: &str) -> OperationError { OperationError { status: StatusCode::BAD_REQUEST, diff --git a/examples/notes/tests/fixtures/ingest-batch.json b/examples/notes/tests/fixtures/ingest-batch.json new file mode 100644 index 0000000..735488e --- /dev/null +++ b/examples/notes/tests/fixtures/ingest-batch.json @@ -0,0 +1,14 @@ +{ + "replay_key": "herdr:cursor:42", + "batch_hash": "sha256:fixture", + "events": [ + { + "type": "workspace_created", + "workspace": "iris" + }, + { + "type": "pane_agent_status_changed", + "status": "working" + } + ] +} diff --git a/examples/notes/tests/surfaces.rs b/examples/notes/tests/surfaces.rs index f2d4f9f..079809f 100644 --- a/examples/notes/tests/surfaces.rs +++ b/examples/notes/tests/surfaces.rs @@ -372,3 +372,91 @@ fn generated_mcp_tool_schema_carries_declared_union() { assert_eq!(one_of[0]["additionalProperties"], json!(false)); assert_eq!(one_of[1]["additionalProperties"], json!(false)); } + +// ── ingest_batch: source-agnostic JSON batch operation (COD-443) ─────────── + +#[tokio::test] +async fn ingest_batch_uses_one_declared_contract_over_http_and_dispatch() { + let state = AppState::with_fixtures(); + let batch: Value = serde_json::from_str(include_str!("fixtures/ingest-batch.json")) + .expect("ingest batch fixture is valid JSON"); + + let direct = via_dispatch(&state, "ingest_batch", json!({"body": batch})).await; + let res = http( + &state, + Request::post("/ingest/batches") + .header("content-type", "application/json") + .body(Body::from(serde_json::to_string(&batch).unwrap())) + .unwrap(), + ) + .await; + assert_eq!(res.status(), StatusCode::OK); + assert_eq!(body_json(res).await, direct); + assert_eq!(direct["accepted"], json!(2)); +} + +#[tokio::test] +async fn generated_cli_and_mcp_expose_the_batch_contract() { + use clap::Parser; + + #[derive(Parser)] + struct Cli { + #[command(subcommand)] + command: notes_example::generated_cli::GeneratedCommand, + } + let cli = Cli::try_parse_from([ + "notes", + "ingest-batch", + "--replay-key", + "herdr:cursor:42", + "--batch-hash", + "sha256:fixture", + "--event", + r#"{"type":"workspace_created"}"#, + "--event", + r#"{"type":"pane_agent_status_changed"}"#, + ]) + .expect("batch flags parse"); + let notes_example::generated_cli::GeneratedCommand::IngestBatch(args) = cli.command else { + panic!("expected ingest-batch subcommand"); + }; + let parameters = + notes_example::generated_cli::GeneratedCommand::IngestBatch(args).parameters_json(); + assert_eq!(parameters["events"].as_array().unwrap().len(), 2); + + // The CLI representation is explicitly repeated JSON strings, so prove + // the one consumer dispatch normalizes it before applying the operation. + let state = AppState::with_fixtures(); + let receipt = execute_operation( + &state, + "ingest_batch", + GeneratedOperationInput { + path: std::collections::BTreeMap::new(), + query: std::collections::BTreeMap::new(), + body: parameters, + }, + ) + .await + .expect("CLI batch dispatch succeeds"); + assert_eq!(receipt["accepted"], json!(2)); + + let mcp: Value = serde_json::from_str(include_str!("../generated/mcp.json")).unwrap(); + let tool = mcp["tools"] + .as_array() + .unwrap() + .iter() + .find(|tool| tool["name"] == json!("ingest_batch")) + .expect("ingest_batch tool present"); + assert_eq!( + tool["inputSchema"]["properties"]["events"]["type"], + json!("array") + ); + assert_eq!( + tool["inputSchema"]["properties"]["events"]["minItems"], + json!(1) + ); + assert_eq!( + tool["inputSchema"]["required"], + json!(["replay_key", "batch_hash", "events"]) + ); +}