refactor(channels): route inbound turns through kernel

extensions/matrix/src/matrix/monitor/handler.test-helpers.ts

@@ -142,6 +142,28 @@ export function createMatrixHandlerTestHarness(
       };
     },
   );
+  const run = vi.fn(
+    async (params: Parameters<MatrixMonitorHandlerParams["core"]["channel"]["turn"]["run"]>[0]) => {
+      const input = await params.adapter.ingest(params.raw);
+      if (!input) {
+        return { admission: { kind: "drop" as const, reason: "ingest-null" }, dispatched: false };
+      }
+      const eventClass = (await params.adapter.classify?.(input)) ?? {
+        kind: "message" as const,
+        canStartAgentTurn: true,
+      };
+      const preflightResult = await params.adapter.preflight?.(input, eventClass);
+      const preflight =
+        preflightResult && "kind" in preflightResult
+          ? { admission: preflightResult }
+          : (preflightResult ?? {});
+      const turn = await params.adapter.resolveTurn(input, eventClass, preflight);
+      if ("runDispatch" in turn) {
+        return await runPrepared(turn);
+      }
+      throw new Error("matrix test helper only supports prepared turn dispatch");
+    },
+  );
   const dmPolicy = options.dmPolicy ?? "open";
   const allowFrom = options.allowFrom ?? (dmPolicy === "open" ? ["*"] : []);
   const cfgForHandler =
@@ -229,6 +251,7 @@ export function createMatrixHandlerTestHarness(
             }),
         },
         turn: {
+          run,
           runPrepared,
         },
         reactions: {

extensions/matrix/src/matrix/monitor/handler.ts

@@ -1829,106 +1829,127 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
           onIdle: typingCallbacks.onIdle,
         });
 
-      const { dispatchResult } = await core.channel.turn.runPrepared({
+      const turnResult = await core.channel.turn.run({
         channel: "matrix",
         accountId: _route.accountId,
-        routeSessionKey: _route.sessionKey,
-        storePath,
-        ctxPayload,
-        recordInboundSession: core.channel.session.recordInboundSession,
-        record: {
-          updateLastRoute: isDirectMessage
-            ? {
-                sessionKey: _route.mainSessionKey,
-                channel: "matrix",
-                to: `room:${roomId}`,
-                accountId: _route.accountId,
+        raw: event,
+        adapter: {
+          ingest: () => ({
+            id: _messageId,
+            rawText: bodyText,
+            textForAgent: ctxPayload.BodyForAgent,
+            textForCommands: ctxPayload.CommandBody,
+            raw: event,
+          }),
+          resolveTurn: () => ({
+            channel: "matrix",
+            accountId: _route.accountId,
+            routeSessionKey: _route.sessionKey,
+            storePath,
+            ctxPayload,
+            recordInboundSession: core.channel.session.recordInboundSession,
+            record: {
+              updateLastRoute: isDirectMessage
+                ? {
+                    sessionKey: _route.mainSessionKey,
+                    channel: "matrix",
+                    to: `room:${roomId}`,
+                    accountId: _route.accountId,
+                  }
+                : undefined,
+              onRecordError: (err) => {
+                logger.warn("failed updating session meta", {
+                  error: String(err),
+                  storePath,
+                  sessionKey: ctxPayload.SessionKey ?? _route.sessionKey,
+                });
+              },
+            },
+            onPreDispatchFailure: () =>
+              core.channel.reply.settleReplyDispatcher({
+                dispatcher,
+                onSettled: () => {
+                  markRunComplete();
+                  markDispatchIdle();
+                },
+              }),
+            runDispatch: async () => {
+              if (
+                sharedDmContextNotice &&
+                markTrackedRoomIfFirst(sharedDmContextNoticeRooms, roomId)
+              ) {
+                client
+                  .sendMessage(roomId, {
+                    msgtype: "m.notice",
+                    body: sharedDmContextNotice,
+                  })
+                  .catch((err) => {
+                    logVerboseMessage(
+                      `matrix: failed sending shared DM session notice room=${roomId}: ${String(err)}`,
+                    );
+                  });
               }
-            : undefined,
-          onRecordError: (err) => {
-            logger.warn("failed updating session meta", {
-              error: String(err),
-              storePath,
-              sessionKey: ctxPayload.SessionKey ?? _route.sessionKey,
-            });
-          },
-        },
-        onPreDispatchFailure: () =>
-          core.channel.reply.settleReplyDispatcher({
-            dispatcher,
-            onSettled: () => {
-              markRunComplete();
-              markDispatchIdle();
+
+              return await core.channel.reply.withReplyDispatcher({
+                dispatcher,
+                onSettled: () => {
+                  markDispatchIdle();
+                },
+                run: async () => {
+                  try {
+                    return await core.channel.reply.dispatchReplyFromConfig({
+                      ctx: ctxPayload,
+                      cfg,
+                      dispatcher,
+                      replyOptions: {
+                        ...replyOptions,
+                        skillFilter: roomConfig?.skills,
+                        // Keep block streaming enabled when explicitly requested, even
+                        // with draft previews on. The draft remains the live preview
+                        // for the current assistant block, while block deliveries
+                        // finalize completed blocks into their own preserved events.
+                        disableBlockStreaming: !blockStreamingEnabled,
+                        onPartialReply: draftStream
+                          ? (payload) => {
+                              latestDraftFullText = payload.text ?? "";
+                              suppressPreviewToolProgressForAnswerText(latestDraftFullText);
+                              updateDraftFromLatestFullText();
+                            }
+                          : undefined,
+                        onBlockReplyQueued: draftStream
+                          ? (payload, context) => {
+                              if (payload.isCompactionNotice === true) {
+                                return;
+                              }
+                              queueDraftBlockBoundary(payload, context);
+                            }
+                          : undefined,
+                        // Reset draft boundary bookkeeping on assistant message
+                        // boundaries so post-tool blocks stream from a fresh
+                        // cumulative payload (payload.text resets upstream).
+                        onAssistantMessageStart: draftStream
+                          ? () => {
+                              resetDraftBlockOffsets();
+                              resetPreviewToolProgress();
+                            }
+                          : undefined,
+                        ...buildPreviewToolProgressReplyOptions(),
+                        onModelSelected,
+                      },
+                    });
+                  } finally {
+                    markRunComplete();
+                  }
+                },
+              });
             },
           }),
-        runDispatch: async () => {
-          if (sharedDmContextNotice && markTrackedRoomIfFirst(sharedDmContextNoticeRooms, roomId)) {
-            client
-              .sendMessage(roomId, {
-                msgtype: "m.notice",
-                body: sharedDmContextNotice,
-              })
-              .catch((err) => {
-                logVerboseMessage(
-                  `matrix: failed sending shared DM session notice room=${roomId}: ${String(err)}`,
-                );
-              });
-          }
-
-          return await core.channel.reply.withReplyDispatcher({
-            dispatcher,
-            onSettled: () => {
-              markDispatchIdle();
-            },
-            run: async () => {
-              try {
-                return await core.channel.reply.dispatchReplyFromConfig({
-                  ctx: ctxPayload,
-                  cfg,
-                  dispatcher,
-                  replyOptions: {
-                    ...replyOptions,
-                    skillFilter: roomConfig?.skills,
-                    // Keep block streaming enabled when explicitly requested, even
-                    // with draft previews on. The draft remains the live preview
-                    // for the current assistant block, while block deliveries
-                    // finalize completed blocks into their own preserved events.
-                    disableBlockStreaming: !blockStreamingEnabled,
-                    onPartialReply: draftStream
-                      ? (payload) => {
-                          latestDraftFullText = payload.text ?? "";
-                          suppressPreviewToolProgressForAnswerText(latestDraftFullText);
-                          updateDraftFromLatestFullText();
-                        }
-                      : undefined,
-                    onBlockReplyQueued: draftStream
-                      ? (payload, context) => {
-                          if (payload.isCompactionNotice === true) {
-                            return;
-                          }
-                          queueDraftBlockBoundary(payload, context);
-                        }
-                      : undefined,
-                    // Reset draft boundary bookkeeping on assistant message
-                    // boundaries so post-tool blocks stream from a fresh
-                    // cumulative payload (payload.text resets upstream).
-                    onAssistantMessageStart: draftStream
-                      ? () => {
-                          resetDraftBlockOffsets();
-                          resetPreviewToolProgress();
-                        }
-                      : undefined,
-                    ...buildPreviewToolProgressReplyOptions(),
-                    onModelSelected,
-                  },
-                });
-              } finally {
-                markRunComplete();
-              }
-            },
-          });
         },
       });
+      if (!turnResult.dispatched) {
+        return;
+      }
+      const { dispatchResult } = turnResult;
       const { queuedFinal, counts } = dispatchResult;
       if (finalReplyDeliveryFailed) {
         if (retryableReplyDeliveryFailed) {