feat(intent): graph writes + automation + summary + batch + single-surface

src/nilscript/dataplane/intent.py

@@ -60,6 +60,7 @@ class Intent:
     about: str                      # ontology type / entity (adapter-agnostic)
     where: tuple[Binding, ...] = ()
     seek: str = "all"               # the | all | count | summary  (read shapes)
+    by: str | None = None           # the grouping dimension for seek="summary"
     change: Change | None = None    # present → a write intent (governed via propose→commit→tier)
     limit: int = 50
     cursor: str | None = None
@@ -132,6 +133,10 @@ def resolve(self, intent: Intent, *, grant_fields: Any = None) -> Outcome:
                     grant_fields=grant_fields,
                 )
                 return Outcome.result(page)
-            return Outcome.refusal("NOT_IMPLEMENTED", "summary (aggregate) lands in the next phase")
+            # seek == "summary": a server-side rollup over a grouping dimension.
+            if not intent.by:
+                return Outcome.refusal("MISSING_DIMENSION", "summary needs a `by` dimension to group on")
+            group_by = self._bind.resolve_attr(intent.about, intent.by)
+            return Outcome.result(self._plane.aggregate(target, filter=filt, group_by=group_by, metrics=("count",)))
         except (ResultTooLarge, CapabilityUnsupported, InvalidFilter) as exc:
             return Outcome.refusal(getattr(exc, "code", "ERROR"), getattr(exc, "message", str(exc)))

src/nilscript/mcp/brain_tools.py

@@ -69,6 +69,30 @@ async def _get(self, path: str, *, params: dict[str, Any]) -> Any:
             if self._client is None:
                 await client.aclose()
 
+    async def _post(self, path: str, *, json: dict[str, Any]) -> Any:
+        client = self._client or httpx.AsyncClient(base_url=self._base, timeout=self._timeout)
+        try:
+            resp = await client.request("POST", path, json=json, headers=self._headers)
+            if resp.status_code >= 400:
+                return {"error": f"brain returned {resp.status_code}", "path": path, "detail": resp.text[:200]}
+            try:
+                return resp.json()
+            except ValueError:
+                return {"error": "non-json response from brain", "status": resp.status_code}
+        except httpx.HTTPError as exc:
+            return {"error": f"brain unreachable: {exc}"}
+        finally:
+            if self._client is None:
+                await client.aclose()
+
+    async def assert_fact(self, event_type: str, facts: dict[str, Any], *, tenant: str | None = None) -> Any:
+        """Assert a business fact to the brain (POST /api/assert) — the brain interprets it via the
+        ontology and executes through the governed kernel path. The write side of a graph intent."""
+        ws = self._tenant_for(tenant)
+        if not ws:
+            return {"error": "no tenant configured (set NIL_BRAIN_TENANT or pass tenant)"}
+        return await self._post("/api/assert", json={"tenant": ws, "event_type": event_type, "facts": facts})
+
     # ── read tools ──────────────────────────────────────────────────────────────────────────────
 
     async def graph(self, kind: str | None = None, tenant: str | None = None) -> dict[str, Any]:

src/nilscript/mcp/server.py

@@ -53,6 +53,7 @@ def build_tools(
     session_id: str = "mcp-session",
     gate: str = "two-step",
     brain: Any = None,
+    automation: Any = None,
 ) -> NilTools:
     """Wire the SDK client to the adapter and wrap it in the MCP tool surface.
 
@@ -67,7 +68,10 @@ def build_tools(
     )
     transport = NilTransport(base_url=adapter_url, bearer_secret=bearer)
     client = NilClient(transport=transport, grant=grant)
-    return NilTools(client, transport, session_id=session_id, gate=gate, brain=brain)
+    return NilTools(
+        client, transport, session_id=session_id, gate=gate,
+        brain=brain, automation=automation, workspace=workspace,
+    )
 
 
 class ToolsProvider:
@@ -174,16 +178,17 @@ def build_server(
         transport_security=transport_security,
     )
 
+    single = _single_surface()  # nil_intent subsumes reads/graph/automation when on
     provider = tools_provider if tools_provider is not None else SingletonToolsProvider(tools)
-    _register_tools(server, provider)
-    if dynamic_verbs:
+    _register_tools(server, provider, single_surface=single)
+    if dynamic_verbs and not single:
         from nilscript.mcp.dynamic import register_dynamic_tools
 
         register_dynamic_tools(server, tools, dynamic_verbs)
     _register_skill(server, tools)
-    if automation_tools is not None:
+    if automation_tools is not None and not single:
         _register_automation_tools(server, automation_tools)
-    if brain_tools is not None:
+    if brain_tools is not None and not single:
         _register_brain_tools(server, brain_tools)
     return server
 
@@ -299,7 +304,13 @@ async def nil_activity(tenant: str | None = None) -> dict[str, Any]:
     )
 
 
-def _register_tools(server: Any, provider: ToolsProvider) -> None:
+def _single_surface() -> bool:
+    """When on, nil_intent is the ONLY model-facing tool (plus describe/commit/status/rollback); the
+    subsumed read/graph/automation tools are hidden. Makes the correct path the only obvious one."""
+    return os.environ.get("NIL_MCP_SINGLE_SURFACE", "") not in ("", "0", "false", "False")
+
+
+def _register_tools(server: Any, provider: ToolsProvider, *, single_surface: bool = False) -> None:
     """Wrap each primitive with the MCP Context so the backend + per-connection session resolve from
     the connection: `provider.get(ctx)` picks the backend (one shared, or per-tenant from headers)
     and `session_key(ctx)` isolates each connection's proposal/idempotency session. `ctx` is injected
@@ -342,24 +353,18 @@ async def nil_status(proposal_id: str, ctx: Context = None) -> dict[str, Any]:
     async def nil_rollback(compensation_token: str, reason: str, ctx: Context = None) -> dict[str, Any]:  # type: ignore[assignment]
         return await provider.get(ctx).rollback(compensation_token, reason, session_id=session_key(ctx))
 
+    # The single-surface keepers: discovery, the one intent payload, and the governance verbs the
+    # approval/reversal flow needs. Everything else is SUBSUMED by nil_intent and hidden when
+    # NIL_MCP_SINGLE_SURFACE is on — so the model sees ONE obvious tool, not a menu.
     server.add_tool(
         nil_describe, name="nil_describe",
         description="Discover the backend skeleton: the verbs and targets it actually exposes. No side effect.",
     )
-    server.add_tool(
-        nil_propose, name="nil_propose",
-        description="Preview an intent (verb + args). NO side effect: returns a human-readable preview "
-        "with a reversibility tier, or a structured refusal. Always call this before nil_commit.",
-    )
     server.add_tool(
         nil_commit, name="nil_commit",
         description="Execute a previously previewed proposal by its id. This is the ONLY tool that writes. "
         "Idempotent: re-committing the same proposal replays, it never double-writes.",
     )
-    server.add_tool(
-        nil_query, name="nil_query",
-        description="Read live business truth (verb + args). No side effect.",
-    )
     server.add_tool(
         nil_intent, name="nil_intent",
         description="THE primary tool. Express WHAT you want as one payload — about (an entity, e.g. "
@@ -374,6 +379,26 @@ async def nil_rollback(compensation_token: str, reason: str, ctx: Context = None
         "Show policies → about='policy', seek='all'. Show business cycles → about='cycle', seek='all'. "
         "Update her phone → about='res.partner', where=[{attr:'name',rel:'contains',value:'دينا'}], change={op:'update', set:{phone:'…'}}.",
     )
+    server.add_tool(
+        nil_status, name="nil_status",
+        description="Get the status/result of a proposal by id, including its compensation handle.",
+    )
+    server.add_tool(
+        nil_rollback, name="nil_rollback",
+        description="Request a governed reversal of a committed effect (compensation_token + reason: "
+        "saga_unwind|owner_cancel|downstream_failed|agent_repair). Previews a compensation to commit, "
+        "or refuses honestly (IRREVERSIBLE / COMPENSATION_EXPIRED). No silent write.",
+    )
+    if single_surface:
+        return  # nil_intent subsumes the rest; hide them so there is ONE obvious tool
+    server.add_tool(
+        nil_propose, name="nil_propose",
+        description="Preview an intent (verb + args). NO side effect: returns a preview + tier, or a refusal.",
+    )
+    server.add_tool(
+        nil_query, name="nil_query",
+        description="Read live business truth (verb + args). No side effect.",
+    )
     server.add_tool(
         nil_search, name="nil_search",
         description="Lean, FILTERED, PAGINATED read of a target (filter=[{field,op,value}], small fields=, "
@@ -397,18 +422,7 @@ async def nil_rollback(compensation_token: str, reason: str, ctx: Context = None
     server.add_tool(
         nil_export, name="nil_export",
         description="Stream a bulk read to a DATA HANDLE (not rows): open it in your sandbox and use code "
-        "(pandas/sqlite) for analysis over many rows. Bulk extraction is gated+audited "
-        "(BULK_APPROVAL_REQUIRED until approved=true).",
-    )
-    server.add_tool(
-        nil_status, name="nil_status",
-        description="Get the status/result of a proposal by id, including its compensation handle.",
-    )
-    server.add_tool(
-        nil_rollback, name="nil_rollback",
-        description="Request a governed reversal of a committed effect (compensation_token + reason: "
-        "saga_unwind|owner_cancel|downstream_failed|agent_repair). Previews a compensation to commit, "
-        "or refuses honestly (IRREVERSIBLE / COMPENSATION_EXPIRED). No silent write.",
+        "(pandas/sqlite) for analysis over many rows. Bulk extraction is gated+audited.",
     )
 
 
@@ -592,12 +606,14 @@ def build_asgi_app(
     verbs: list[str] = []
     if dynamic_tools and not multi_tenant:
         verbs = asyncio.run(_discover_verbs(adapter_url, bearer))
+    from nilscript.mcp.automation_tools import AutomationTools
     from nilscript.mcp.brain_tools import BrainTools
 
     brain = BrainTools.from_env()  # graph/meta domain behind nil_intent (None if NIL_BRAIN_URL unset)
+    automation = AutomationTools.from_env()  # automation domain behind nil_intent
     tools = build_tools(
         adapter_url=adapter_url, grant_id=grant_id, workspace=workspace,
-        bearer=bearer, scopes=scopes, gate=gate, brain=brain,
+        bearer=bearer, scopes=scopes, gate=gate, brain=brain, automation=automation,
     )
     provider: ToolsProvider | None = None
     if multi_tenant:
@@ -611,11 +627,9 @@ def build_asgi_app(
             default=default, allow_insecure=allow_insecure, gate=gate,
             registry=make_registry_lookup(),
         )
-    from nilscript.mcp.automation_tools import AutomationTools
-
     server = build_server(
         tools, dynamic_verbs=verbs, tools_provider=provider,
-        automation_tools=AutomationTools.from_env(),
+        automation_tools=automation,
         brain_tools=brain,
         allowed_hosts=_allowed_hosts_from_env(),
     )

src/nilscript/mcp/tools.py

@@ -79,6 +79,8 @@ def __init__(
         session_id: str = "mcp-session",
         gate: str = "two-step",
         brain: Any = None,
+        automation: Any = None,
+        workspace: str = "",
     ) -> None:
         if gate not in GATE_MODES:
             raise ValueError(f"gate must be one of {sorted(GATE_MODES)}, got {gate!r}")
@@ -87,6 +89,8 @@ def __init__(
         self._default_session = session_id
         self._gate = gate
         self._brain = brain  # optional BrainTools — owns graph/meta entities in nil_intent routing
+        self._automation = automation  # optional AutomationTools — owns about="automation"
+        self._workspace = workspace
         self._proposals: dict[str, dict[str, dict[str, Any]]] = {}
 
     def _sid(self, session_id: str | None) -> str:
@@ -170,6 +174,7 @@ async def intent(
         where: list[dict[str, Any]] | None = None,
         seek: str = "all",
         change: dict[str, Any] | None = None,
+        by: str | None = None,
         limit: int = 50,
         cursor: str | None = None,
         *,
@@ -184,11 +189,14 @@ async def intent(
             about=about,
             where=tuple(Binding(attr=b.get("attr"), rel=b.get("rel"), value=b.get("value")) for b in (where or [])),
             seek=seek,
+            by=by,
             change=Change(op=change.get("op"), set=change.get("set") or {}) if change else None,
             limit=limit,
             cursor=cursor,
         )
         providers: list[Any] = []
+        if self._automation is not None:
+            providers.append(_AutomationIntentProvider(self._automation, self._workspace))
         if self._brain is not None:
             providers.append(_GraphIntentProvider(self._brain))
         providers.append(_AdapterIntentProvider(self, session_id))  # fallback: owns business entities
@@ -207,10 +215,28 @@ async def _adapter_resolve(self, intent_obj: Any, session_id: str | None) -> dic
                 session_id=session_id,
             )
         return await self.query("nil.intent", {
-            "about": intent_obj.about, "where": where, "seek": intent_obj.seek,
+            "about": intent_obj.about, "where": where, "seek": intent_obj.seek, "by": intent_obj.by,
             "limit": intent_obj.limit, "cursor": intent_obj.cursor,
         })
 
+    async def intent_batch(
+        self, intents: list[dict[str, Any]], *, session_id: str | None = None
+    ) -> dict[str, Any]:
+        """Run many intents in one call — each resolved independently (partial-allow): one intent's
+        refusal never drops the others. The system executes 100% of what's permitted, structurally."""
+        results: list[dict[str, Any]] = []
+        for spec in intents:
+            try:
+                outcome = await self.intent(
+                    spec.get("about", ""), spec.get("where"), spec.get("seek", "all"),
+                    spec.get("change"), spec.get("by"), int(spec.get("limit") or 50), spec.get("cursor"),
+                    session_id=session_id,
+                )
+            except Exception as exc:  # noqa: BLE001 — isolate a bad intent as its own refusal
+                outcome = {"outcome": "refused", "code": "ERROR", "message": str(exc)}
+            results.append({"about": spec.get("about", ""), "result": outcome})
+        return {"outcome": "batch", "count": len(results), "results": results}
+
     async def _intent_change(
         self, about: str, where: list[dict[str, Any]], change: dict[str, Any], *, session_id: str | None
     ) -> dict[str, Any]:
@@ -363,6 +389,12 @@ def owns(self, about: str) -> bool:
 
     async def resolve(self, intent: Any) -> Outcome:
         a = intent.about
+        if intent.change is not None:  # graph write → assert a fact; the brain interprets + governs
+            facts = dict(intent.change.set or {})
+            for b in intent.where:
+                facts[b.attr] = b.value
+            data = await self._brain.assert_fact(f"{a}.{intent.change.op}", facts)
+            return Outcome.result(data)
         if a in ("cycle", "cycles"):
             data = await self._brain.cycles()
         elif a == "overview":
@@ -389,3 +421,28 @@ def owns(self, about: str) -> bool:
 
     async def resolve(self, intent: Any) -> Outcome:
         return Outcome.result(await self._tools._adapter_resolve(intent, self._session_id))
+
+
+class _AutomationIntentProvider:
+    """The automation domain: about="automation" resolves against the automation registry — reads list
+    the workspace's automations, a create registers a new one. Same provider contract, no keywords."""
+
+    def __init__(self, automation: Any, workspace: str) -> None:
+        self._a = automation
+        self._ws = workspace
+
+    def owns(self, about: str) -> bool:
+        return about == "automation"
+
+    async def resolve(self, intent: Any) -> Outcome:
+        if intent.change is not None:
+            c = intent.change
+            if c.op != "create":
+                return Outcome.refusal("UNSUPPORTED_OP", f"automation supports create (register); got {c.op!r}")
+            s = c.set or {}
+            data = await self._a.register(
+                s.get("automation_id") or s.get("id", ""), s.get("name") or {},
+                s.get("plan") or {}, s.get("trigger") or {},
+            )
+            return Outcome.result(data)
+        return Outcome.result(await self._a.list(self._ws))

tests/test_intent_reads.py

@@ -49,12 +49,17 @@ def get_one(self, target, record_id, fields):
         return next((r for r in self.rows if r["id"] == record_id), None)
 
     def aggregate(self, target, *, predicates, group_by, metrics):
-        return None
+        buckets: dict = {}
+        for r in self.rows:
+            if self._match(r, predicates):
+                buckets[r.get(group_by)] = buckets.get(r.get(group_by), 0) + 1
+        return [{"key": k, "count": v} for k, v in buckets.items()]
 
 
 def _contacts():
-    rows = [{"id": i, "name": f"Contact {i}", "phone": f"+9745{i:07d}"} for i in range(40)]
-    rows.append({"id": 18, "name": "دينا كمال النجار", "phone": "+97455123456"})
+    rows = [{"id": i, "name": f"Contact {i}", "phone": f"+9745{i:07d}", "country": ("QA" if i % 2 else "SA")}
+            for i in range(40)]
+    rows.append({"id": 18, "name": "دينا كمال النجار", "phone": "+97455123456", "country": "QA"})
     return rows
 
 
@@ -94,3 +99,16 @@ def test_unknown_about_is_a_structured_refusal() -> None:
     out = _resolver().resolve(Intent(about="hr.salary", where=(), seek="count"))
     assert out.kind == "refusal"
     assert out.code  # carries a code the agent can act on
+
+
+def test_seek_summary_groups_via_aggregate() -> None:
+    intent = Intent(about="res.partner", where=(), seek="summary", by="country")
+    out = _resolver().resolve(intent)
+    assert out.kind == "result"
+    by = {g["key"]: g["count"] for g in out.value["groups"]}
+    assert by == {"SA": 20, "QA": 21}
+
+
+def test_seek_summary_without_a_dimension_is_a_refusal() -> None:
+    out = _resolver().resolve(Intent(about="res.partner", where=(), seek="summary"))
+    assert out.kind == "refusal" and out.code == "MISSING_DIMENSION"