feat: add JSON batch operation reference (COD-443)

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 '<json object>'` 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

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

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<Vec<String>>,
 }
 
+#[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<Vec<String>>,
+}
+

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<crate::AppState> {
@@ -57,6 +58,7 @@ pub fn generated_router() -> Router<crate::AppState> {
         .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<crate::AppState>,
+    Json(body): Json<Value>,
+) -> Response {
+    crate::execute_operation_http(
+        &state,
+        "ingest_batch",
+        GeneratedOperationInput {
+            path: BTreeMap::new(),
+            query: BTreeMap::new(),
+            body,
+        },
+    )
+    .await
+}

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"
     }
   ]
 }

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<Vec<Value>, 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<Value, OperationError> {
+    #[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 <json>` values to the ordered JSON objects
+/// HTTP and MCP callers pass directly in the declared batch field.
+fn normalize_batch_events(events: &Value) -> Result<Vec<Value>, 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,

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"
+    }
+  ]
+}

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"])
+    );
+}