fix(pgwire): answer kv writes with one INSERT tag

nodedb/src/control/server/response_shape/types/plan_kind.rs

@@ -151,7 +151,10 @@ pub fn describe_plan(plan: &PhysicalPlan) -> PlanKind {
         PhysicalPlan::Document(DocumentOp::PointPut { .. })
         | PhysicalPlan::Document(DocumentOp::PointInsert { .. })
         | PhysicalPlan::Document(DocumentOp::BatchInsert { .. })
+        | PhysicalPlan::Kv(KvOp::Insert { .. })
         | PhysicalPlan::Kv(KvOp::InsertIfAbsent { .. })
+        | PhysicalPlan::Kv(KvOp::Put { .. })
+        | PhysicalPlan::Kv(KvOp::BatchPut { .. })
         | PhysicalPlan::Columnar(ColumnarOp::Insert { .. }) => DmlResult("INSERT"),
 
         PhysicalPlan::Document(DocumentOp::PointUpdate {

nodedb/src/data/executor/handlers/kv/crud/write_basic.rs

@@ -80,7 +80,10 @@ impl CoreLoop {
             // stored post-image, not an echo of the request.
             return self.kv_stored_returning_response(task, spec, rls_filters, &[(key, value)]);
         }
-        self.response_ok(task)
+        // A put always writes its row, so the statement affected exactly one:
+        // the tag is `INSERT 0 1`, never a bare `OK` (pgwire's generic tag for
+        // a plan that reports nothing).
+        self.response_affected(task, 1)
     }
 
     /// SQL `INSERT` semantics: write only if key doesn't already exist.
@@ -167,7 +170,10 @@ impl CoreLoop {
         if let Some(spec) = returning {
             return self.kv_stored_returning_response(task, spec, rls_filters, &[(key, value)]);
         }
-        self.response_ok(task)
+        // An insert writes exactly one row or fails the statement, so the
+        // affected count is 1 and pgwire renders `INSERT 0 1` — the same tag
+        // the document engine's point insert produces.
+        self.response_affected(task, 1)
     }
 
     /// SQL `INSERT ... ON CONFLICT DO NOTHING` semantics: write if absent,

nodedb/tests/wire/cases/sql_dml_affected_counts.rs

@@ -286,6 +286,52 @@ async fn kv_multi_key_delete_reports_matched_key_count() {
     );
 }
 
+/// A key-value `INSERT` of a new key writes exactly one row, so its command tag
+/// must report `1`. A kv write that answers with no affected count renders as a
+/// bare `OK` tag, which the client's tag parser reads as `0` rows — the count is
+/// the only signal distinguishing "wrote" from "matched nothing".
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn kv_insert_of_new_key_reports_one() {
+    let server = TestServer::start().await;
+    server
+        .exec("CREATE COLLECTION kv_probe (key TEXT PRIMARY KEY, n INT) WITH (engine='kv')")
+        .await
+        .unwrap();
+
+    let count = affected(&server, "INSERT INTO kv_probe (key, n) VALUES ('a', 1)").await;
+    assert_eq!(
+        count, 1,
+        "a KV INSERT that wrote one key must report 1, not a bare tag the client reads as 0"
+    );
+    assert_eq!(
+        live_rows(&server, "SELECT count(*) FROM kv_probe WHERE key = 'a'").await,
+        1,
+        "the key must really be present, so the reported count is the honest one"
+    );
+}
+
+/// A key-value `UPSERT` overwrites or writes exactly one key, so its tag must
+/// report `1` for the same reason: the PUT path always writes the key.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn kv_upsert_of_new_key_reports_one() {
+    let server = TestServer::start().await;
+    server
+        .exec("CREATE COLLECTION kv_probe (key TEXT PRIMARY KEY, n INT) WITH (engine='kv')")
+        .await
+        .unwrap();
+
+    let count = affected(&server, "UPSERT INTO kv_probe (key, n) VALUES ('a', 1)").await;
+    assert_eq!(
+        count, 1,
+        "a KV UPSERT that wrote one key must report 1, not a bare tag the client reads as 0"
+    );
+    assert_eq!(
+        live_rows(&server, "SELECT count(*) FROM kv_probe WHERE key = 'a'").await,
+        1,
+        "the upserted key must really be present"
+    );
+}
+
 /// A CRDT document collection routes its PK-targeted delete through the CRDT
 /// engine, which shares the count contract: a delete that removed the row
 /// reports `1`.