fix(cli): consolidate provider + KB probe-gates into one strict shared helper

docs/mcp/tools/knowledge-bases.md

@@ -77,6 +77,7 @@ The toolkit validates the definition client-side because the API accepts shapes
 - A green `validate_knowledge_base_access` proves **read access only** (`read_ai_agents` on the pipe) — never the `manage_ai_agents` entitlement that every knowledge base create / update / delete (plain text, document, data lookup) requires.
 - An **empty knowledge base list** is a valid green result (`knowledge_base_count: 0`), not a failure.
 - The **CLI gates writes** on the probe: `pipefy kb plain-text create` / `update`, `pipefy kb document create` / `update`, and `pipefy kb data-lookup create` / `update` run the read-access probe first and fail with the classified problem if it is denied, before attempting the mutation.
+- **Clean-gate contract.** The probe can return a success that still carries a `problem` — when the API returns partial data alongside GraphQL errors, it surfaces the classified error rather than discarding it, and deliberately does not flip `ok`. The CLI write gate therefore treats the gate as clean only when it is **`ok` and carries no `problem`**; a present `problem` is partial denial and is never read as full access.
 - The **MCP tools stay explicit-validate-first**: create / update do not auto-probe. Call `validate_knowledge_base_access` yourself before writing.
 - **Deletes require confirmation**: the MCP tool needs `confirm=true`; the CLI needs `--yes` (or an interactive prompt).
 

packages/cli/src/pipefy_cli/commands/_common.py

@@ -320,6 +320,20 @@ def merge_extra_attrs(
     return merged
 
 
+def probe_gate(probe: dict[str, Any]) -> dict[str, Any] | None:
+    """Gate a write on an access-probe result; return a failure dict or None.
+
+    Clean only when the probe is ``ok`` **and** carries no ``problem``: a probe
+    can return ``ok: true`` with a non-null ``problem`` when the API returns
+    partial data alongside GraphQL errors, and that partial denial must never be
+    read as full access. A non-clean gate returns the classified problem so the
+    write never runs (the CLI then exits 1 via ``exit_1_on_unsuccessful``).
+    """
+    if probe.get("ok") and "problem" not in probe:
+        return None
+    return {"success": False, **probe}
+
+
 def confirm_destructive(*, yes: bool, description: str, verb: str = "delete") -> None:
     """Prompt before a destructive action unless ``yes`` is True."""
     if yes:

packages/cli/src/pipefy_cli/commands/ai_provider.py

@@ -15,7 +15,11 @@
 import typer
 from pipefy_sdk import PipefyClient
 
-from pipefy_cli.commands._common import confirm_destructive, run_cli_command
+from pipefy_cli.commands._common import (
+    confirm_destructive,
+    probe_gate,
+    run_cli_command,
+)
 
 ai_provider_app = typer.Typer(
     help="LLM providers (organization-scoped: discovery reads and provider writes).",
@@ -35,21 +39,6 @@
 )
 
 
-async def _provider_probe_gate(client: PipefyClient, org_uuid: str) -> dict | None:
-    """Gate a write on the read-access probe; return a failure dict or None.
-
-    Treats the gate as clean only when the probe is ``ok`` **and** carries no
-    ``problem``: a probe can return ``ok: true`` with a non-null ``problem`` when
-    the API returns partial data alongside GraphQL errors, and that partial denial
-    must never be read as full access. A non-clean gate returns the classified
-    problem so the write never runs and the CLI exits 1.
-    """
-    probe = await client.validate_llm_provider_access(org_uuid)
-    if probe.get("ok") and "problem" not in probe:
-        return None
-    return {"success": False, **probe}
-
-
 @ai_provider_app.command("list")
 def ai_provider_list(
     ctx: typer.Context,
@@ -207,7 +196,7 @@ def ai_provider_create(
     """
 
     async def factory(client: PipefyClient):
-        gate = await _provider_probe_gate(client, org_uuid)
+        gate = probe_gate(await client.validate_llm_provider_access(org_uuid))
         if gate is not None:
             return gate
         provider = await client.create_llm_provider(
@@ -244,7 +233,7 @@ def ai_provider_update(
     """
 
     async def factory(client: PipefyClient):
-        gate = await _provider_probe_gate(client, org_uuid)
+        gate = probe_gate(await client.validate_llm_provider_access(org_uuid))
         if gate is not None:
             return gate
         provider = await client.update_llm_provider(

packages/cli/src/pipefy_cli/commands/knowledge_base.py

@@ -11,6 +11,7 @@
 from pipefy_cli.commands._common import (
     confirm_destructive,
     parse_json_value,
+    probe_gate,
     run_cli_command,
 )
 
@@ -145,7 +146,7 @@ def kb_plain_text_create(
     """
 
     async def factory(client: PipefyClient):
-        gate = await _probe_gate(client, pipe_uuid)
+        gate = probe_gate(await client.validate_knowledge_base_access(pipe_uuid))
         if gate is not None:
             return gate
         plain_text = await client.create_ai_knowledge_base_plain_text(
@@ -181,7 +182,7 @@ def kb_plain_text_update(
     """
 
     async def factory(client: PipefyClient):
-        gate = await _probe_gate(client, pipe_uuid)
+        gate = probe_gate(await client.validate_knowledge_base_access(pipe_uuid))
         if gate is not None:
             return gate
         plain_text = await client.update_ai_knowledge_base_plain_text(
@@ -278,7 +279,7 @@ def kb_document_create(
     """
 
     async def factory(client: PipefyClient):
-        gate = await _probe_gate(client, pipe_uuid)
+        gate = probe_gate(await client.validate_knowledge_base_access(pipe_uuid))
         if gate is not None:
             return gate
         try:
@@ -324,7 +325,7 @@ def kb_document_update(
     """
 
     async def factory(client: PipefyClient):
-        gate = await _probe_gate(client, pipe_uuid)
+        gate = probe_gate(await client.validate_knowledge_base_access(pipe_uuid))
         if gate is not None:
             return gate
         document = await client.update_ai_knowledge_base_document(
@@ -452,7 +453,7 @@ def kb_data_lookup_create(
     fields, condition_list = _parse_data_lookup_options(output_fields, conditions)
 
     async def factory(client: PipefyClient):
-        gate = await _probe_gate(client, pipe_uuid)
+        gate = probe_gate(await client.validate_knowledge_base_access(pipe_uuid))
         if gate is not None:
             return gate
         data_lookup = await client.create_ai_knowledge_base_data_lookup(
@@ -504,7 +505,7 @@ def kb_data_lookup_update(
     fields, condition_list = _parse_data_lookup_options(output_fields, conditions)
 
     async def factory(client: PipefyClient):
-        gate = await _probe_gate(client, pipe_uuid)
+        gate = probe_gate(await client.validate_knowledge_base_access(pipe_uuid))
         if gate is not None:
             return gate
         data_lookup = await client.update_ai_knowledge_base_data_lookup(
@@ -548,15 +549,3 @@ async def factory(client: PipefyClient):
         )
 
     run_cli_command(ctx, json_out, factory, exit_1_on_unsuccessful=True)
-
-
-async def _probe_gate(client: PipefyClient, pipe_uuid: str) -> dict | None:
-    """Gate a write on the read-access probe; return a failure dict or None.
-
-    A failed probe returns the classified problem so the write never runs and
-    the CLI exits 1 (via ``exit_1_on_unsuccessful``) with the problem rendered.
-    """
-    probe = await client.validate_knowledge_base_access(pipe_uuid)
-    if probe.get("ok"):
-        return None
-    return {"success": False, **probe}

packages/cli/tests/test_knowledge_base_commands.py

@@ -217,6 +217,46 @@ def test_kb_plain_text_create_denied_probe_blocks_write(
     mock_client.create_ai_knowledge_base_plain_text.assert_not_awaited()
 
 
+def test_kb_plain_text_create_blocked_by_partial_denial_probe(
+    runner, clean_pipefy_env, saved_cwd, monkeypatch
+):
+    """A probe that is ok BUT carries a problem is partial denial: write blocked."""
+    _env(monkeypatch)
+    mock_client = MagicMock()
+    mock_client.validate_knowledge_base_access = AsyncMock(
+        return_value={
+            "ok": True,
+            "knowledge_base_count": 1,
+            "problem": {"kind": "permission_denied", "message": "Partial denial"},
+        }
+    )
+    mock_client.create_ai_knowledge_base_plain_text = AsyncMock()
+
+    with _client_patch(mock_client):
+        result = runner.invoke(
+            app,
+            [
+                "kb",
+                "plain-text",
+                "create",
+                "--pipe-uuid",
+                "pipe-uuid-1",
+                "--name",
+                "n",
+                "--content",
+                "c",
+                "--description",
+                "d",
+                "--json",
+            ],
+        )
+    assert result.exit_code == 1
+    data = json.loads(result.stdout)
+    assert data["success"] is False
+    assert data["problem"]["kind"] == "permission_denied"
+    mock_client.create_ai_knowledge_base_plain_text.assert_not_awaited()
+
+
 def test_kb_plain_text_update_partial(runner, clean_pipefy_env, saved_cwd, monkeypatch):
     _env(monkeypatch)
     mock_client = MagicMock()