fix(channels): the host posts nothing into a channel its server declared inbound-only

changelog.d/inbound-only-channels.fixed.md

@@ -0,0 +1,38 @@
+- The host no longer publishes into a channel its server declared inbound-only
+  (`direction: "inbound"`, SPEC §14.2). A server's live-transcript channel can
+  be one, where an agent speaking in the room arrives addressed. Until now
+  that moved a resident's plain speech there, the delivery attempt opened the
+  channel if she had closed it, and the server refused the publish.
+  - Every publish of plain speech into such a channel is refused before the
+    channel is opened or published to, a drafts resend and `{{unsent}}`
+    included, with a reason that names the declaration and says to use that
+    server's own tools. A channel declared inbound-only while the publish
+    waits, on the channel's lifecycle work, its open or anything else, is
+    refused too, and so is one whose registration is gone by the time the
+    words would be published. Nothing is published to either. A declaration
+    that comes with the channel's own opening leaves the channel open: the
+    agent gets the failure for its words, and the notice that the channel is
+    open if this open is what opened it.
+  - A `>>` or `>>>` target naming the channel bounces with the same reason,
+    and the closed-channel invitation offers no reply into it. In explicit
+    mode the refused target also clears the turn's destination, so the turn's
+    later text without a prefix bounces too, rather than going to the
+    destination named before it. A delivery refused because the channel was
+    declared inbound-only after its target resolved ends the destination
+    too, unless a later round has named another channel by then.
+  - `channel_publish` and the subconscious's `speak_in_channel` return an
+    error that names it. The outgoing stream skips the channel, and the
+    channel context names no default outgoing channel for it.
+  - A channel the host knows only from a push is refused once the server's
+    answer to opening it, wherever the host opens it, declares it
+    inbound-only.
+  - The routing notices say plain speech is refused there, and to use that
+    server's own tools. The locus and hybrid mode primers, shown when an
+    agent's mode changes, say plain speech goes to the locus unless its server
+    declared that channel inbound-only.
+- `speak_in_channel`'s `mcpl:speech-routed` trace now follows a confirmed
+  delivery, with `serverId`, `delivered` and `textLen` as plain speech's trace
+  has them, and no `messageId`, as before. Any other result is traced as
+  `mcpl:speech-route-failed`, with the publish's reason, the connector's
+  detail when it sent some, and `outcome: "unknown"` when the words may have
+  been posted or `"failed"` when nothing was, a refusal included.

docs/explicit-prose-routing.md

@@ -26,7 +26,9 @@ A prose segment's FIRST line may carry a routing prefix:
   unprefixed segments of the SAME turn, including those a later stream of it
   writes, such as a framework retry's, a budget restart's or the compile a
   provider-capacity hold defers the turn to. A round kept private (one that
-  calls `skip_reply`, or has a same-round private think) sets no target.
+  calls `skip_reply`, or has a same-round private think) sets no target. A
+  target its server declared inbound-only is refused and clears the turn's
+  target, so later unprefixed segments bounce until a new `>>` target.
   Cross-turn, nothing is implicit.
 - **Continuation modifier:** ` !` immediately after the target
   (`>>#ops !` / `>>skip_reply !`) requests an immediate re-wake after this turn
@@ -81,7 +83,7 @@ depends on the route:
 
 Explicit send tools (`send_message` etc.) are unchanged. Channel lifecycle
 invariants (send-opens-channel, subscribed⇒open) are unchanged — a `>>` send
-into a closed channel opens it, with the usual `[channels]` notice.
+into a channel decided closed opens it, with the usual `[channels]` notice.
 
 ## Teaching the grammar (teach-by-bounce)
 

src/mcpl/prose-stream-router.ts

@@ -16,8 +16,13 @@
  *
  * Suppression is fail-closed: text whose destination is unknown (explicit
  * mode before the first `>>`, unresolvable/ambiguous targets, `>>skip_reply`)
- * emits NOTHING — the delivery path will bounce/retain it; a stream consumer
- * must never see text that delivery would refuse to send.
+ * emits NOTHING — the delivery path will bounce/retain it. Text for a channel
+ * the `speakable` option refuses emits nothing either, and which servers see
+ * what is emitted is the registry's (sendOutgoingChunk). The stream can't
+ * know what delivery decides later: `speakable` is asked as the text is
+ * routed, so a channel declared inbound-only after its text was emitted, or
+ * one whose open then fails, is refused at delivery, though a consumer has
+ * seen the text.
  *
  * Three modes, mirroring proseRouting:
  *  - explicit: parse `>>` prefixes. A fresh turn's stream starts suppressed;
@@ -58,6 +63,14 @@ export class ProseStreamRouter {
       initialTarget: string | null;
       /** Resolve a `>>spec` to a channelId, or null (unresolved/ambiguous). */
       resolve: (spec: string) => string | null;
+      /** Whether speech may be published to this channel now. Text for a
+       *  channel it refuses is neither emitted nor kept for outgoing/complete,
+       *  so the stream's end can't carry it there should that change before
+       *  the stream ends. Asked as the text is routed, while delivery decides
+       *  when its round is admitted: text refused here stays out of the
+       *  stream even if the channel accepts speech again by then and the
+       *  round posts it. Every channel, when left out. */
+      speakable?: (channelId: string) => boolean;
     },
   ) {
     this.target = opts.initialTarget;
@@ -74,7 +87,8 @@ export class ProseStreamRouter {
    * this at each round's end, with the turn's destination: a private round's
    * prefixes, a `>>skip_reply` envelope and a target that didn't resolve all
    * leave that destination as it was, though the stream followed them while
-   * the round was written.
+   * the round was written. A target refused for its server's inbound-only
+   * declaration clears it.
    */
   retarget(target: string | null): void {
     this.target = target;
@@ -100,7 +114,7 @@ export class ProseStreamRouter {
   feed(delta: string): RoutedDelta[] {
     const out: RoutedDelta[] = [];
     for (const ch of delta) this.step(ch, out);
-    return coalesce(out);
+    return this.keep(out);
   }
 
   /**
@@ -127,7 +141,7 @@ export class ProseStreamRouter {
     this.held = '';
     this.token = '';
     if (this.opts.mode !== 'locus') this.state = 'line-start';
-    return coalesce(out);
+    return this.keep(out);
   }
 
   /** End of stream: the last segment ends. */
@@ -253,7 +267,17 @@ export class ProseStreamRouter {
   private emit(text: string, out: RoutedDelta[]): void {
     if (this.target === null || !text) return;
     out.push({ channelId: this.target, delta: text });
-    this.accumulated.set(this.target, (this.accumulated.get(this.target) ?? '') + text);
+  }
+
+  /**
+   * What one feed or segment end routed, less any channel `speakable`
+   * refuses, kept for outgoing/complete. A feed runs without yielding, so
+   * one answer per channel holds for all of it.
+   */
+  private keep(out: RoutedDelta[]): RoutedDelta[] {
+    const kept = coalesce(out).filter((d) => this.opts.speakable?.(d.channelId) ?? true);
+    for (const d of kept) this.accumulated.set(d.channelId, (this.accumulated.get(d.channelId) ?? '') + d.delta);
+    return kept;
   }
 }
 

src/types/trace.ts

@@ -442,13 +442,16 @@ export type TraceEvent =
     })
 
   // Host-owned speech routing (ChannelRegistry.routeSpeech). One trace per
-  // delivered prose segment. `text` is the segment verbatim — the same content
+  // delivered prose segment, and one per publish the subconscious's
+  // speak_in_channel delivers (ChannelRegistry.publishForAgent): that one's
+  // conversationId is the speaking agent's name, and it carries no
+  // messageId. `text` is the segment verbatim — the same content
   // already on the bus chunk-by-chunk via inference:tokens, and mcpl:* traces
   // are 'ops'-scoped on the webui wire — and `messageId` is the surface's id
-  // for the posted message (when the MCPL server reports one). Together they
-  // give external taps the (channelId, messageId, text) tuple needed to edit
-  // a just-posted message, e.g. a TTS interruption truncating a reply to the
-  // words actually voiced.
+  // for the posted message (on routeSpeech's traces, when the MCPL server
+  // reports one). Together they give external taps the (channelId,
+  // messageId, text) tuple needed to edit a just-posted message, e.g. a TTS
+  // interruption truncating a reply to the words actually voiced.
   | (TraceEventBase & {
       type: 'mcpl:speech-routed';
       /** Agent / conversation whose speech was routed. */
@@ -459,7 +462,8 @@ export type TraceEvent =
       textLen: number;
       /** The routed prose segment, verbatim. */
       text: string;
-      /** Surface message id (e.g. Discord message id), when reported. */
+      /** Surface message id (e.g. Discord message id), when reported, on
+       *  routeSpeech's traces. speak_in_channel's carry none. */
       messageId?: string;
     })
   | (TraceEventBase & {
@@ -473,6 +477,9 @@ export type TraceEvent =
       /** `failed`: nothing was posted; `unknown`: the request was dispatched
        *  and no valid receipt came back, so it may have been. */
       outcome?: 'failed' | 'unknown';
+      /** The connector error's structured data, bounded (connectorText),
+       *  when it sent some: it may say what a multi-part send posted. */
+      detail?: string;
     })
 
   // Admin puppet: an operator executed a tool AS an agent and stored the

test/channel-lifecycle-order.test.ts

@@ -40,14 +40,20 @@ function fixture(onNotice?: NoticeHandler, store?: JsStore, coalesce?: RegistryO
   const routeFailures: Array<{ reason: string }> = [];
   // A server that answers an open for another channel, by requested id.
   const answeredFor = new Map<string, string>();
+  // The direction an open's answer declares, by the address the open was
+  // sent to; an address not here declares the descriptor's own.
+  const declares = new Map<string, ChannelDescriptor['direction']>();
   const serve = (kind: 'open' | 'close', params: { channelId?: string; address?: unknown }) => {
     const id = params.channelId!;
     calls.push({ kind, id, address: params.address });
     return new Promise<any>((resolve, reject) => {
       const finish = () => {
         const target = kind === 'open' ? answeredFor.get(id) ?? id : id;
         actual.set(target, kind === 'open');
-        resolve(kind === 'open' ? { channel: descriptor(target) } : { closed: true });
+        const declared = declares.get(String((params.address as { value?: unknown } | undefined)?.value));
+        resolve(kind === 'open'
+          ? { channel: { ...descriptor(target), ...(declared ? { direction: declared } : {}) } }
+          : { closed: true });
       };
       if (held.has(id)) pending.push({ finish, fail: () => reject(new Error('fixture lifecycle failure')) });
       else finish();
@@ -87,7 +93,7 @@ function fixture(onNotice?: NoticeHandler, store?: JsStore, coalesce?: RegistryO
     assert.equal(settled, true, 'lifecycle operations must settle');
     await result;
   };
-  return { registry, actual, calls, pending, held, traces, isOpen, tool, drain, publishedWhileOpen, notices, routeFailures, replaceServer, answeredFor, setGrant };
+  return { registry, actual, calls, pending, held, traces, isOpen, tool, drain, publishedWhileOpen, notices, routeFailures, replaceServer, answeredFor, declares, setGrant };
 }
 
 test('remove/re-add during reconcile converges on the replacement desired state, with ACK first (#185)', async () => {
@@ -317,6 +323,108 @@ test('a draft published into a channel with a close pending waits for it and pos
   assert.equal(f.notices.length, 0);
 });
 
+test('a channel declared inbound-only while a publish waits for its lifecycle work is refused, and nothing is published', async () => {
+  // A publish that finds the channel's lifecycle work queued waits for it.
+  // A channels/changed during that wait can declare the channel
+  // inbound-only, so the publish checks the channel again after the wait,
+  // just before it publishes, and refuses it there.
+  const f = fixture();
+  await f.registry.handleChanged('test', { added: [descriptor('x')] });
+  f.held.add('x');
+  const open = f.tool('open');
+  await tick();
+  const draft = f.registry.publish('resident', 'a draft', { serverId: 'test', channelId: 'x' });
+  await tick();
+  await f.registry.handleChanged('test', { updated: [{ ...descriptor('x'), direction: 'inbound' }] });
+  await f.drain(open, draft);
+  assert.equal((await open).success, true);
+  const outcome = await draft;
+  assert.equal(outcome.status, 'failed');
+  assert.match(outcome.reason ?? '', /^inbound-only: server "test" declared this channel's direction "inbound"/);
+  assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published');
+  assert.equal(f.isOpen(), true, 'the open the agent asked for stands');
+});
+
+for (const change of ['announced again as inbound-only', 'removed'] as const) {
+  test(`a channel ${change} while its opening is announced is refused, and nothing is published`, async () => {
+    // The notice of an opening runs after the open and before the publish. A
+    // channels/changed there can replace the channel's entry or remove it,
+    // so the check before the publish reads the channel's current entry, not
+    // the one the open returned.
+    let announced: Promise<unknown> | undefined;
+    let f!: ReturnType<typeof fixture>;
+    f = fixture(() => {
+      announced = f.registry.handleChanged('test', change === 'removed'
+        ? { removed: ['x'] }
+        : { added: [{ ...descriptor('x'), direction: 'inbound' }] });
+    });
+    await f.registry.handleChanged('test', { added: [descriptor('x')] });
+    assert.equal(await f.registry.routeSpeech('resident', 'hello', 'x'), null);
+    assert.equal(f.notices.length, 1, 'the opening was announced');
+    assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published');
+    assert.match(f.routeFailures[0]?.reason ?? '', change === 'removed'
+      ? /^the channel was removed before the words were published$/
+      : /^inbound-only: server "test" declared this channel's direction "inbound"/);
+    await announced;
+  });
+}
+
+for (const opener of ['tool', 'reconcile'] as const) {
+  test(`an open answered after the agent closed the channel during it leaves it closed, and its inbound-only declaration applies (${opener})`, async () => {
+    // The close is the newer decision, so the channel ends closed. The answer
+    // is still the server's word about the channel, so a later publish is
+    // refused before anything opens.
+    const f = fixture();
+    f.declares.set('original', 'inbound');
+    if (opener === 'tool') await f.registry.handleChanged('test', { added: [descriptor('x')] });
+    f.held.add('x');
+    const opening = opener === 'tool'
+      ? f.tool('open')
+      : f.registry.handleChanged('test', { added: [descriptor('x', true)] });
+    await tick();
+    assert.equal(f.pending.length, 1, 'the open is in flight');
+    const close = f.tool('close');
+    await f.drain(opening, close);
+    assert.equal((await close).success, true);
+    assert.equal(f.registry.getDesiredState('test', 'x'), 'closed');
+    assert.equal(f.isOpen(), false);
+    assert.equal(f.actual.get('x'), false);
+    assert.equal(f.registry.inboundOnlyServer('x', 'test'), 'test', "the late answer's declaration applies");
+    const opens = f.calls.filter((c) => c.kind === 'open').length;
+    const outcome = await f.registry.publish('resident', 'words', { serverId: 'test', channelId: 'x' });
+    assert.equal(outcome.status, 'failed');
+    assert.match(outcome.reason ?? '', /^inbound-only: /);
+    assert.equal(f.calls.filter((c) => c.kind === 'open').length, opens, 'refused before anything opens');
+  });
+}
+
+test("an open answered for another channel applies nothing of its answer to the channel opened", async () => {
+  // A substitution fails the open before the answer's declaration is read.
+  // That is what lets an answer whose channel names no id stand for the
+  // channel opened.
+  const f = fixture();
+  f.answeredFor.set('x', 'y');
+  f.declares.set('original', 'inbound');
+  await f.registry.handleChanged('test', { added: [descriptor('x')] });
+  assert.equal((await f.tool('open')).success, false);
+  assert.ok(f.traces.some((e) => e.type === 'mcpl:channel-open-substituted' && e.channelId === 'x'),
+    'the open reached the server and was answered for another channel');
+  assert.equal(f.registry.inboundOnlyServer('x', 'test'), undefined, "the other channel's declaration isn't applied");
+});
+
+test("an open's answer for a target the channel has since left applies nothing; the new target's answer decides", async () => {
+  const f = fixture();
+  f.declares.set('original', 'inbound');
+  f.held.add('x');
+  const opening = f.registry.handleChanged('test', { added: [descriptor('x', true)] });
+  await tick();
+  await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'new-target')] });
+  await f.drain(opening);
+  assert.deepEqual(f.calls.map((c) => c.address), [{ value: 'original' }, { value: 'new-target' }]);
+  assert.equal(f.isOpen(), true);
+  assert.equal(f.registry.inboundOnlyServer('x', 'test'), undefined, "the old target's declaration isn't applied");
+});
+
 test('waiting for a backscroll does not issue new opens or auto-open notices for each speech', async () => {
   const f = fixture();
   await f.registry.handleChanged('test', { added: [descriptor('x', true)] });