diff --git a/docs/mcp/tools/knowledge-bases.md b/docs/mcp/tools/knowledge-bases.md index 88454d30..b5e50557 100644 --- a/docs/mcp/tools/knowledge-bases.md +++ b/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). diff --git a/packages/cli/src/pipefy_cli/commands/_common.py b/packages/cli/src/pipefy_cli/commands/_common.py index 0171fb61..bd1979f0 100644 --- a/packages/cli/src/pipefy_cli/commands/_common.py +++ b/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: diff --git a/packages/cli/src/pipefy_cli/commands/ai_provider.py b/packages/cli/src/pipefy_cli/commands/ai_provider.py index e84542d2..ec887405 100644 --- a/packages/cli/src/pipefy_cli/commands/ai_provider.py +++ b/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( diff --git a/packages/cli/src/pipefy_cli/commands/knowledge_base.py b/packages/cli/src/pipefy_cli/commands/knowledge_base.py index 0e2bb0b9..e56bc348 100644 --- a/packages/cli/src/pipefy_cli/commands/knowledge_base.py +++ b/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} diff --git a/packages/cli/tests/test_knowledge_base_commands.py b/packages/cli/tests/test_knowledge_base_commands.py index de4efbc5..32030109 100644 --- a/packages/cli/tests/test_knowledge_base_commands.py +++ b/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()