Merge pull request #76 from shetty4l/fix/output-routing-and-cleanups

feat: add built-in tools, output routing, and idleTimeout

Author: shetty4l

bun.lock

@@ -19,7 +19,7 @@
     "prepare": "husky"
   },
   "dependencies": {
-    "@shetty4l/core": "^0.1.30"
+    "@shetty4l/core": "^0.1.33"
   },
   "devDependencies": {
     "@biomejs/biome": "^2.3.13",

package.json

@@ -67,6 +67,9 @@ export interface CortexConfig {
   // Thalamus
   thalamusModel: string;
   thalamusSyncIntervalMs: number;
+
+  // Output routing
+  silentChannelAlias?: string;
 }
 
 // --- Defaults ---
@@ -150,6 +153,7 @@ function validateConfig(raw: unknown): Result<Partial<CortexConfig>> {
     { key: "telegramBotToken", label: "telegramBotToken" },
     { key: "systemPromptFile", label: "systemPromptFile" },
     { key: "thalamusModel", label: "thalamusModel" },
+    { key: "silentChannelAlias", label: "silentChannelAlias" },
   ];
 
   for (const field of stringFields) {

src/config.ts

@@ -21,6 +21,8 @@ import { startServer } from "./server";
 import { createEmptyRegistry, loadSkills, type SkillRegistry } from "./skills";
 import { Thalamus } from "./thalamus";
 import { Tick } from "./tick";
+import { type BuiltinToolContext, createCombinedRegistry } from "./tools";
+import { createSendMessageTool } from "./tools/send-message";
 import { VERSION } from "./version";
 
 const log = createLogger("cortex");
@@ -81,10 +83,23 @@ export async function startCortexRuntime(
   const server = deps.startServer(config, thalamus);
   deps.log(`listening on http://${config.host}:${config.port}`);
 
-  const loop = deps.startProcessingLoop(config, registry);
+  // Create channel registry (needed by built-in tools)
+  const channels = deps.createChannelRegistry(config, thalamus);
+
+  // Create built-in tools with mutable per-message context
+  const builtinCtx: BuiltinToolContext = { topicKey: "" };
+  const builtinTools = [createSendMessageTool(channels)];
+  const combinedRegistry = createCombinedRegistry(
+    builtinTools,
+    registry,
+    () => builtinCtx,
+  );
+
+  const loop = deps.startProcessingLoop(config, combinedRegistry, {
+    builtinContext: builtinCtx,
+  });
   deps.log("processing loop started");
 
-  const channels = deps.createChannelRegistry(config, thalamus);
   await channels.startAll();
 
   await thalamus.start();

src/index.ts

@@ -31,6 +31,8 @@ import { buildPrompt, loadAndRenderSystemPrompt } from "./prompt";
 import type { SkillRegistry } from "./skills";
 import type { OpenAITool } from "./synapse";
 import { chat } from "./synapse";
+import type { BuiltinToolContext } from "./tools";
+import { resolveOutputChannel } from "./tools";
 
 // --- Constants ---
 
@@ -67,6 +69,8 @@ export interface ProcessingLoopOptions {
   pollBusyMs?: number;
   /** Override idle poll interval (ms). Default: 2000. */
   pollIdleMs?: number;
+  /** Mutable context shared with built-in tools (topicKey updated per message). */
+  builtinContext?: BuiltinToolContext;
 }
 
 /**
@@ -85,6 +89,7 @@ export function startProcessingLoop(
   let running = true;
   const pollBusyMs = options?.pollBusyMs ?? DEFAULT_POLL_BUSY_MS;
   const pollIdleMs = options?.pollIdleMs ?? DEFAULT_POLL_IDLE_MS;
+  const builtinCtx = options?.builtinContext;
 
   if (!config.extractionModel) {
     log("extraction disabled — no extractionModel configured");
@@ -119,6 +124,11 @@ export function startProcessingLoop(
           delay = pollBusyMs;
           const startMs = performance.now();
 
+          // Update built-in tool context for this message
+          if (builtinCtx) {
+            builtinCtx.topicKey = message.topic_key;
+          }
+
           const preview =
             message.text.length > 60
               ? `${message.text.slice(0, 57)}...`
@@ -232,7 +242,7 @@ export function startProcessingLoop(
 
             // 8. Write to outbox
             enqueueOutboxMessage({
-              channel: message.channel,
+              channel: resolveOutputChannel(message.channel, config),
               topicKey: message.topic_key,
               text: responseText,
             });

src/loop.ts

@@ -351,6 +351,7 @@ export function startServer(
     port: config.port,
     host: config.host,
     version: VERSION,
+    idleTimeout: 120,
     onRequest: async (req: Request, url: URL) => {
       const start = performance.now();
       let response: Response | null = null;

src/server.ts

@@ -203,7 +203,7 @@ export class Thalamus {
     // Create inbox messages for each triaged group
     for (const item of items) {
       const idempotencyHash = [...item.rawBufferIds].sort().join(",");
-      const idempotencyKey = `thalamus-sync:${idempotencyHash}`;
+      const idempotencyKey = `thalamus-sync:${item.topicKey}:${idempotencyHash}`;
 
       enqueueInboxMessage({
         channel: "thalamus",

src/thalamus/index.ts

@@ -0,0 +1,97 @@
+/**
+ * Built-in tools for Cortex.
+ *
+ * Built-in tools are first-party primitives that run inside the agent loop
+ * with direct access to cortex internals (outbox, channels, config).
+ * They are NOT namespaced — tool names are bare (e.g. "send_message").
+ *
+ * External skills loaded from skillDirs are namespaced as "skillId.toolName".
+ */
+
+import type { Result } from "@shetty4l/core/result";
+import type { CortexConfig } from "../config";
+import type { SkillRegistry, SkillToolResult, ToolDefinition } from "../skills";
+
+// --- Types ---
+
+/** Mutable context updated per-message in the processing loop. */
+export interface BuiltinToolContext {
+  /** Topic key of the message currently being processed. */
+  topicKey: string;
+}
+
+/** A built-in tool with direct access to cortex internals. */
+export interface BuiltinTool {
+  readonly definition: ToolDefinition;
+  execute(
+    argsJson: string,
+    ctx: BuiltinToolContext,
+  ): Promise<Result<SkillToolResult>>;
+}
+
+// --- Output channel resolution ---
+
+/** Channels that are internal (system) and should not receive user-facing responses. */
+const SYSTEM_CHANNELS = new Set(["thalamus", "calendar"]);
+
+/**
+ * Resolve the output channel for a response.
+ *
+ * - If the input channel is a system channel, route to "silent".
+ * - If the channel is "silent" and an alias is configured, redirect to the alias.
+ * - Otherwise, return the channel unchanged (echo back to sender).
+ */
+export function resolveOutputChannel(
+  inputChannel: string,
+  config: Pick<CortexConfig, "silentChannelAlias">,
+): string {
+  const channel = SYSTEM_CHANNELS.has(inputChannel) ? "silent" : inputChannel;
+  if (channel === "silent" && config.silentChannelAlias) {
+    return config.silentChannelAlias;
+  }
+  return channel;
+}
+
+// --- Combined registry ---
+
+/**
+ * Create a unified SkillRegistry that dispatches to built-in tools first,
+ * then falls through to the external skill registry.
+ *
+ * Built-in tools receive a BuiltinToolContext (topicKey, etc.) instead of
+ * the SkillRuntimeContext used by external skills.
+ *
+ * @param builtinTools Array of built-in tool definitions + executors.
+ * @param skillRegistry External skill registry loaded from skillDirs.
+ * @param getContext Closure that returns the current per-message context.
+ */
+export function createCombinedRegistry(
+  builtinTools: BuiltinTool[],
+  skillRegistry: SkillRegistry,
+  getContext: () => BuiltinToolContext,
+): SkillRegistry {
+  const builtinMap = new Map(builtinTools.map((t) => [t.definition.name, t]));
+
+  const allTools: ReadonlyArray<ToolDefinition> = [
+    ...builtinTools.map((t) => t.definition),
+    ...skillRegistry.tools,
+  ];
+
+  return {
+    tools: allTools,
+
+    async executeTool(name, argumentsJson, ctx) {
+      const builtin = builtinMap.get(name);
+      if (builtin) {
+        return builtin.execute(argumentsJson, getContext());
+      }
+      return skillRegistry.executeTool(name, argumentsJson, ctx);
+    },
+
+    isMutating(name) {
+      const builtin = builtinMap.get(name);
+      if (builtin) return builtin.definition.mutatesState;
+      return skillRegistry.isMutating(name);
+    },
+  };
+}

src/tools/index.ts

@@ -0,0 +1,72 @@
+/**
+ * send_message built-in tool.
+ *
+ * Allows the agent to proactively send a message to the user on a
+ * specific channel. Validates the channel exists and can deliver,
+ * then writes to the outbox for async delivery.
+ */
+
+import { err, ok } from "@shetty4l/core/result";
+import type { ChannelRegistry } from "../channels";
+import { enqueueOutboxMessage } from "../db";
+import type { BuiltinTool, BuiltinToolContext } from "./index";
+
+export function createSendMessageTool(
+  channelRegistry: ChannelRegistry,
+): BuiltinTool {
+  return {
+    definition: {
+      name: "send_message",
+      description:
+        "Send a message to the user on a specific channel (e.g. 'telegram'). Use this to proactively share information.",
+      inputSchema: {
+        type: "object",
+        properties: {
+          channel: {
+            type: "string",
+            description: "Target delivery channel (e.g. 'telegram')",
+          },
+          text: {
+            type: "string",
+            description: "Message text to send",
+          },
+        },
+        required: ["channel", "text"],
+      },
+      mutatesState: true,
+    },
+
+    async execute(argsJson: string, ctx: BuiltinToolContext) {
+      const args = JSON.parse(argsJson) as {
+        channel?: string;
+        text?: string;
+      };
+
+      if (!args.channel || typeof args.channel !== "string") {
+        return err("channel is required and must be a string");
+      }
+      if (!args.text || typeof args.text !== "string") {
+        return err("text is required and must be a string");
+      }
+
+      // Validate channel exists and can deliver
+      const channel = channelRegistry.get(args.channel);
+      if (!channel) {
+        return err(`unknown channel: '${args.channel}'`);
+      }
+      if (!channel.canDeliver) {
+        return err(`channel '${args.channel}' cannot deliver messages`);
+      }
+
+      enqueueOutboxMessage({
+        channel: args.channel,
+        topicKey: ctx.topicKey,
+        text: args.text,
+      });
+
+      return ok({
+        content: `Message queued for delivery on ${args.channel}`,
+      });
+    },
+  };
+}