channel_egress: send through hestia_channel_send (hestia #1372, shadow)

sage/gateway/being_gate_client.py

@@ -2678,8 +2678,13 @@ def _unbounded_reason(effector: str, args: Optional[dict] = None) -> str:
                      "script, use memory_edit (start_line 1 to the last line replaces all of it). "
                      "To start fresh, write to a new file name.",
                      {"path": "path to your note", "content": "what to write"}, ["path", "content"]),
-    "channel_egress": ("Send a message out through a sealed channel.",
-                       {"to": "recipient", "body": "your message"}, ["to", "body"]),
+    "channel_egress": ("Send a message out through a channel the operator bound for you. You name the channel; "
+                       "who it reaches is set by the operator's binding, not by you. Asking for a channel you do not "
+                       "have answers with the ones you do. Today every send is a SHADOW: judged, witnessed and kept "
+                       "in your outbox, and NOT delivered; the receipt says so. Once sends are delivered, a fourth "
+                       "unanswered message on one channel within the same hour will not go.",
+                       {"channel": "the id of a channel bound for you", "body": "your message"},
+                       ["channel", "body"]),
     "mesh": ("Wake another member through the fractal mesh with a pointer-based notice "
              "(no body — point at content you already posted).",
              {"to": "member name", "kind": "notice kind, e.g. coordination, reply, ack",

sage/gateway/hestia_dispatch.py

@@ -11,8 +11,8 @@
     peer_ask         none — a COMPOSE: publish the question at a         EXECUTED as the compose
                        pointer, member_notify(kind=coordination),
                        drain hestia_member_inbox for the answer
-    channel_egress   hestia_egress_pending is read-side only;           honest `pending`
-                       no send-side tool exists
+    channel_egress   hestia_channel_send(channel_id, content,           EXECUTED, in shadow:
+                       session_id); the binding names the recipient       judged, kept, not delivered
 
 Result envelope (r1, pinned from the owner seat): {ok, result|error, witness_id}. When the
 failure originates in the daemon, `error` carries hestia's own `hestia.<code>` string as its
@@ -4391,16 +4391,39 @@ def _wake_addressee(self, to: str, meta: dict, turn: dict) -> Optional[str]:
             return ", ".join(woke)
         return None
 
-    # -- channel_egress: not built on the daemon ----------------------------
+    # -- channel_egress: hestia_channel_send (hestia #1344 slice 1, SHADOW) -----
     def _do_channel_egress(self, intent: BeingIntent) -> ResultEnvelope:
-        # SAY WHAT IT IS, NOT WHAT IT AWAITS. The note named design work ("F5 enforcement gated on rate-governor
-        # calibration, PRD §12"), and sprout-being took it for a setting it could complete and asked dp what
-        # "rate-governor calibration" required (room seq 306, 2026-10-19), having tried to run shell scripts by
-        # egressing them to stdout. The design state stays in this comment; the being gets the fact.
-        return ResultEnvelope(ok=False, pending=True,
-                              note="channel_egress is not built yet. Nothing was sent, whatever you give it, and "
-                                   "there is no step or setting that makes it work today. It sends messages out; "
-                                   "it never runs anything. To have a script run, request_run asks your seat.")
+        """A message to a channel the operator bound for the being (hestia #1372).
+
+        THE BEING NAMES A CHANNEL, NEVER A RECIPIENT. Who a channel reaches is the operator's
+        binding; the daemon builds the gate target from it and never from caller text. So `to`
+        (the old schema's word) is read as the channel's id, and nothing here turns a name or an
+        address into one.
+
+        SHADOW IS SAID, NOT IMPLIED. The daemon judges, witnesses and keeps every send and
+        delivers none until the delivery slice. A receipt that reads like a sent message would
+        be the misleading kind (sprout-being took the old stub's design words for a setting it
+        could complete, room seq 306): `sent: false` and the daemon's own note go through as they
+        are, with the budget."""
+        channel = str(intent.args.get("channel") or intent.args.get("to") or "").strip()
+        body = str(intent.args.get("body") or intent.args.get("content") or "")
+        if not channel or not body.strip():
+            from sage.gateway.reference_f1a import missing_args
+            return ResultEnvelope(ok=False, error=missing_args(intent.args, ("channel", "body"), "channel_egress"))
+        out = self._call("hestia_channel_send", {"channel_id": channel, "content": body})
+        err = _hestia_error(out)
+        if err:
+            if "unknown_tool" in err:
+                # A daemon from before hestia #1372: say it, rather than a bare code.
+                return ResultEnvelope(ok=False, error=(
+                    "channel_egress: this machine's hestia has no channel send yet, so nothing was sent or "
+                    "recorded. It arrives with a hestia deploy; there is nothing for you to do."))
+            return ResultEnvelope(ok=False, error=err)
+        return ResultEnvelope(ok=True, witness_id=out.get("witnessEntryHash"), result={
+            "sent": bool(out.get("sent")), "shadow": bool(out.get("shadow")),
+            "channel": out.get("channel_id"), "recipient_class": out.get("recipient_class"),
+            "outbox_id": out.get("outbox_id"), "decision": out.get("decision"),
+            "budget": out.get("budget"), "note": out.get("note")})
 
 
 def _fd_path(fd: int, fallback: str) -> str:

sage/gateway/tests/test_hestia_dispatch.py

@@ -286,13 +286,73 @@ def test_drain_inbox_returns_notices():
     assert [a for n, a in FakeMcp.calls if n == "hestia_member_inbox"][-1]["session_id"] == "sid-1"
 
 
-def test_channel_egress_pending_not_built():
-    d, _ = _disp()
-    env = d(BeingIntent("channel_egress", {"to": "x", "body": "y"}), _ALLOW)
-    assert not env.ok and env.pending and "not built yet" in env.note
-    # the being gets the fact, not design jargon it can mistake for a setting (room seq 306, 2026-10-19)
-    assert "calibration" not in env.note and "no step or setting" in env.note
-    assert "never runs anything" in env.note
+class _Channels(FakeMcp):
+    """hestia_channel_send as #1372's daemon answers it: one bound channel, shadow, nothing sent."""
+    known = True
+
+    def call(self, name, args):
+        if name != "hestia_channel_send":
+            return super().call(name, args)
+        FakeMcp.calls.append((name, args))
+        if not _Channels.known:
+            body = {"_hestia_error": {"code": "hestia.unknown_tool", "message": "Unknown tool: hestia_channel_send"}}
+        elif args.get("channel_id") != "dp-email":
+            body = {"_hestia_error": {"code": "hestia.channel_unbound", "message":
+                    f"no channel '{args.get('channel_id')}' is bound for you. Your bound channels: dp-email. "
+                    "Nothing was sent or recorded."}, "bound_channels": ["dp-email"]}
+        else:
+            body = {"shadow": True, "sent": False, "outbox_id": 4, "channel_id": "dp-email",
+                    "recipient_class": "operator", "decision": "would_send",
+                    "budget": {"limit": 3, "window": "1h", "unanswered": 0, "would_deny": False},
+                    "witnessEntryHash": "chan-hash",
+                    "note": "Shadow: nothing left this machine. Once sends are enforced this would go to "
+                            "'dp-email' (operator). It is in your outbox as #4."}
+        return {"result": {"structuredContent": body}}
+
+
+def _cdisp():
+    FakeMcp.calls, FakeMcp.fail, _Channels.known = [], {}, True
+    root = tempfile.mkdtemp(prefix="hd-")
+    return HestiaF1aDispatcher("sprout-being", root, mcp_factory=lambda ep, pid: _Channels(ep, pid)), root
+
+
+def test_channel_egress_sends_through_the_daemon_and_says_it_was_a_shadow():
+    """hestia #1372: judged, witnessed and kept, not delivered. The receipt must not read as a sent message."""
+    d, _ = _cdisp()
+    env = d(BeingIntent("channel_egress", {"channel": "dp-email", "body": "a test message"}), _ALLOW)
+    assert env.ok and env.witness_id == "chan-hash", env
+    assert env.result["sent"] is False and env.result["shadow"] is True and env.result["outbox_id"] == 4
+    assert env.result["note"].startswith("Shadow: nothing left this machine")
+    assert env.result["budget"]["limit"] == 3
+    (sent,) = _mb_calls("hestia_channel_send")
+    # the channel id and the text, with the being's own session; never a recipient or an address
+    assert sent == {"channel_id": "dp-email", "content": "a test message", "session_id": "sid-1"}
+
+
+def test_channel_egress_reads_the_old_to_as_the_channel_id_never_as_a_recipient():
+    d, _ = _cdisp()
+    env = d(BeingIntent("channel_egress", {"to": "dp-email", "body": "hi"}), _ALLOW)
+    assert env.ok and _mb_calls("hestia_channel_send")[-1]["channel_id"] == "dp-email"
+    env = d(BeingIntent("channel_egress", {"to": "dp", "body": "hi"}), _ALLOW)
+    assert not env.ok and env.error.startswith("hestia.channel_unbound") and "dp-email" in env.error, env
+
+
+def test_channel_egress_on_a_daemon_without_the_tool_says_so_plainly():
+    d, _ = _cdisp()
+    _Channels.known = False
+    env = d(BeingIntent("channel_egress", {"channel": "dp-email", "body": "hi"}), _ALLOW)
+    assert not env.ok and "no channel send yet" in env.error and "nothing for you to do" in env.error
+    # the old stub's design words never come back (room seq 306, 2026-10-19)
+    assert "calibration" not in env.error
+
+
+def test_channel_egress_without_a_channel_or_body_asks_nothing_of_the_daemon():
+    d, _ = _cdisp()
+    env = d(BeingIntent("channel_egress", {"channel": "dp-email", "body": "  "}), _ALLOW)
+    assert not env.ok and "body" in env.error
+    env = d(BeingIntent("channel_egress", {"body": "hi"}), _ALLOW)
+    assert not env.ok and "channel" in env.error
+    assert _mb_calls("hestia_channel_send") == []
 
 
 def test_local_verbs_delegate_to_reference():