n8n Agent Builder: ห้าม Node ใน Workflow แล้ว Agent ยังเรียก Tool ได้หรือไม่?

"""Original teaching MOCK, Python 3.12; NOT n8n code or an n8n policy resolver.

No network, credentials, external writes or real tool execution. Tests NOT RUN.
The AUTHOR'S stricter missing-context refusal is not n8n's module/license posture.
"""
from dataclasses import dataclass
import json
from typing import Callable, Mapping


class Stopped(Exception):
    pass


@dataclass(frozen=True)
class PolicySnapshot:
    revision: str
    blocked_types: frozenset[str]


@dataclass(frozen=True)
class ToolDefinition:
    canonical_type: str
    fixture_result: str


class PolicyGateMock:
    """Synchronous single-worker fixture; deliberately no rule inheritance/cache."""

    def __init__(
        self,
        registry: Mapping[str, ToolDefinition],
        current_policy: Callable[[], PolicySnapshot | None],
    ) -> None:
        self._registry = dict(registry)  # Host-owned, never supplied by the model.
        self._current_policy = current_policy
        self.executed: list[str] = []
        self.ledger: list[dict[str, str]] = []

    def _record(self, tool_id: str, node_type: str, revision: str, reason: str) -> None:
        # No user request, arguments, result, token or credential in this ledger.
        self.ledger.append({
            "tool_id": tool_id,
            "canonical_type": node_type,
            "policy_revision": revision,
            "decision": reason,
        })

    def _read_snapshot(self) -> PolicySnapshot:
        snapshot = self._current_policy()
        if snapshot is None:
            raise Stopped("AUTHOR_POLICY_UNAVAILABLE")
        if not isinstance(snapshot.revision, str) or not snapshot.revision.strip():
            raise Stopped("AUTHOR_REVISION_MISSING")
        if not isinstance(snapshot.blocked_types, frozenset) or any(
            not isinstance(name, str) or not name.strip()
            for name in snapshot.blocked_types
        ):
            raise Stopped("AUTHOR_POLICY_INVALID")
        return snapshot

    def _definition(self, tool_id: str) -> ToolDefinition:
        if not isinstance(tool_id, str) or tool_id not in self._registry:
            raise Stopped("AUTHOR_UNKNOWN_TOOL")
        definition = self._registry[tool_id]
        if not isinstance(definition.canonical_type, str) or not definition.canonical_type.strip():
            raise Stopped("AUTHOR_TYPE_MISSING")
        return definition

    def preflight(self, tool_ids: list[str]) -> str:
        """Author's strict advisory check; no save grandfathering is modelled."""
        snapshot = self._read_snapshot()
        for tool_id in tool_ids:
            if self._definition(tool_id).canonical_type in snapshot.blocked_types:
                raise Stopped("AUTHOR_PREFLIGHT_BLOCKED")
        return snapshot.revision  # Evidence of that check, not an execution permit.

    def call(self, identity: str, tool_id: str) -> str:
        revision, node_type = "UNKNOWN", "UNKNOWN"
        try:
            if not isinstance(identity, str) or not identity.strip():
                raise Stopped("AUTHOR_IDENTITY_MISSING")
            definition = self._definition(tool_id)
            node_type = definition.canonical_type
            # Fresh provider read on EACH call, even if preflight was clear.
            snapshot = self._read_snapshot()
            revision = snapshot.revision
            if node_type in snapshot.blocked_types:
                raise Stopped("AUTHOR_TYPE_BLOCKED")
        except Stopped as error:
            safe_id = tool_id if isinstance(tool_id, str) and tool_id in self._registry else "UNKNOWN"
            self._record(safe_id, node_type, revision, str(error))
            raise
        # Only this in-memory fixture runs. This is NOT execute()/supplyData() in n8n.
        self.executed.append(tool_id)
        self._record(tool_id, node_type, revision, "AUTHOR_FIXTURE_ALLOWED")
        return definition.fixture_result


def demo() -> dict[str, object]:
    holder = {"policy": PolicySnapshot("p1", frozenset())}
    gate = PolicyGateMock({
        "mail_summary": ToolDefinition("mailTool", "fixture mail summary"),
        "notes_read": ToolDefinition("notesTool", "fixture note"),
    }, lambda: holder["policy"])
    preflight_revision = gate.preflight(["mail_summary", "notes_read"])
    holder["policy"] = PolicySnapshot("p2", frozenset({"mailTool"}))
    outcomes = []
    for tool_id in ["mail_summary", "notes_read"]:
        try:
            outcomes.append(gate.call("fixture-reader", tool_id))
        except Stopped as error:
            outcomes.append(str(error))
    return {
        "mode": "SYNTHETIC_MOCK",
        "preflight_revision": preflight_revision,
        "outcomes": outcomes,
        "executed": gate.executed,
        "ledger": gate.ledger,
    }


if __name__ == "__main__":
    print(json.dumps(demo(), indent=2, ensure_ascii=False))

"""Eight authored synthetic unit tests. NOT RUN. Python standard library only."""
import unittest
from policy_gate_mock import PolicyGateMock, PolicySnapshot, Stopped, ToolDefinition


class PolicyGateTests(unittest.TestCase):
    def setUp(self):
        self.holder = {"policy": PolicySnapshot("p1", frozenset())}
        self.registry = {
            "mail": ToolDefinition("mailTool", "fixture mail"),
            "notes": ToolDefinition("notesTool", "fixture note"),
        }
        self.gate = PolicyGateMock(self.registry, lambda: self.holder["policy"])

    def test_changed_policy_after_preflight_prevents_fixture(self):
        self.assertEqual(self.gate.preflight(["mail"]), "p1")
        self.holder["policy"] = PolicySnapshot("p2", frozenset({"mailTool"}))
        with self.assertRaisesRegex(Stopped, "AUTHOR_TYPE_BLOCKED"):
            self.gate.call("reader", "mail")
        self.assertEqual(self.gate.executed, [])
        self.assertEqual(self.gate.ledger[-1]["policy_revision"], "p2")

    def test_allowed_fixture_only(self):
        self.assertEqual(self.gate.call("reader", "notes"), "fixture note")
        self.assertEqual(self.gate.executed, ["notes"])

    def test_unknown_tool_stops(self):
        with self.assertRaisesRegex(Stopped, "AUTHOR_UNKNOWN_TOOL"):
            self.gate.call("reader", "model_invented")
        self.assertEqual(self.gate.executed, [])

    def test_missing_identity_stops(self):
        with self.assertRaisesRegex(Stopped, "AUTHOR_IDENTITY_MISSING"):
            self.gate.call(" ", "notes")
        self.assertEqual(self.gate.executed, [])

    def test_missing_policy_stops_under_author_policy(self):
        self.holder["policy"] = None
        with self.assertRaisesRegex(Stopped, "AUTHOR_POLICY_UNAVAILABLE"):
            self.gate.call("reader", "notes")
        self.assertEqual(self.gate.executed, [])

    def test_missing_revision_stops_under_author_policy(self):
        self.holder["policy"] = PolicySnapshot("", frozenset())
        with self.assertRaisesRegex(Stopped, "AUTHOR_REVISION_MISSING"):
            self.gate.call("reader", "notes")
        self.assertEqual(self.gate.executed, [])

    def test_friendly_tool_id_cannot_override_registry_type(self):
        gate = PolicyGateMock({
            "friendly_notes": ToolDefinition("mailTool", "fixture only")
        }, lambda: PolicySnapshot("p2", frozenset({"mailTool"})))
        with self.assertRaisesRegex(Stopped, "AUTHOR_TYPE_BLOCKED"):
            gate.call("reader", "friendly_notes")
        self.assertEqual(gate.executed, [])

    def test_ledger_has_only_fixed_decision_fields(self):
        self.gate.call("reader", "notes")
        self.assertEqual(set(self.gate.ledger[-1]), {
            "tool_id", "canonical_type", "policy_revision", "decision"
        })
        self.assertNotIn("fixture note", str(self.gate.ledger))


if __name__ == "__main__":
    unittest.main()

# Policy boundary teaching fixture

Status: **NOT RUN**. Neither the example nor these eight synthetic tests was executed,
imported or compiled during editorial production. This is original teaching code,
not an n8n SDK, not an implementation of its policy resolver and not production middleware.

- Runtime: Python 3.12 or newer, standard library only.
- Dependencies: none; no pip installation is required.
- Environment variables: none. Never add a real token to this fixture.
- Installation: copy `policy_gate_mock.py` and `test_policy_gate_mock.py` into one local directory.
- Run, only when authorized: `python policy_gate_mock.py`.
- Input: the in-file trusted registry, initial policy `p1`, then policy `p2` blocking `mailTool`.
- Expected semantic output: `mode=SYNTHETIC_MOCK`, preflight `p1`, outcomes
  `AUTHOR_TYPE_BLOCKED` and `fixture note`, executed list contains only `notes_read`.
  Both decisions in the ledger use current revision `p2`.
- Test instructions, only when authorized: `python -m unittest -v test_policy_gate_mock.py`.
  Eight synthetic cases are authored; no pass count is reported because they were not run.
- Permissions: local memory and stdout only. No network, database, credentials, files written,
  actual node execution or external effects. Tool names and results are invented fixtures.
- Cost: no external provider/API fee in this fixture; local execution resources remain the user's.

## What the example teaches

An advisory preflight at `p1` is evidence of that moment, not a reusable execution permit.
The call resolves the model's tool ID through a host-owned registry and reads a policy
snapshot again immediately before its in-memory fixture. A friendly tool ID cannot supply
an alternative type. A blocked call never reaches `executed`.

Missing identity, unknown tool, missing policy or revision stops under **the author's strict
sample policy**. This deliberately differs from the inspected n8n feature: missing module /
unregistered check allows, and an expired license stops enforcement. Do not use this fixture
to infer n8n's fail posture.

## Limits

Synchronous single worker only. There is no policy store, distributed invalidation, license,
module loader, type variant normalization, project/instance intersection, rule priority,
credential rule, save grandfathering, deadline, transport authentication, authorization of
the supplied identity, atomic policy-and-execute transaction, concurrency control or rollback.
The string identity is an input-presence demonstration, not an authenticated identity.
The provider may change policy immediately after the read; no zero-race claim is made.
Unexpected provider exceptions propagate without executing the fixture but do not produce
the structured refusal ledger. That is a limitation to address before building a service.
The ledger is an in-memory list, not persistent/tamper-resistant audit evidence.

## Separate n8n live verification plan — NOT RUN

Have an authorized administrator inspect the installed version, release channel, relevant
license feature and `type-availability-policies` module on every main/worker process. Inspect
the actual policy and project without editing production settings. The article's source
baseline is `n8n@2.44.0`; it is a prerelease, not a recommendation to upgrade production.
Read current official release notes before deciding compatibility.

In an independently authorized, isolated test environment, design cases for Agent Builder
node tools: existing blocked type can save; publish/start denies it; an allowed preflight
followed by a changed rule is rechecked at tool execution; unknown/dynamic inline contents
are not claimed statically covered; main/worker configuration and license-loss posture are
observed explicitly. Check execute and supplyData paths, policy versions/cache propagation,
error records and whether an actual fixture-side counter remains untouched after refusal.
Do not call real write tools merely to prove denial. Use disposable read-only fixtures and
capture the real results before claiming integration PASS.

This editorial run did not install/upgrade n8n, call its API, modify rules, start workflows,
run nodes or contact any external service from this example.