getcompanion-ai/co-mono
1
0
Fork 0
mirror of https://github.com/getcompanion-ai/co-mono.git synced 2026-04-19 17:04:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

new gateway

Browse source
This commit is contained in:
Harivansh Rathi 2026-03-05 18:58:27 -08:00
parent 01958298e0
commit 9a0b848789
34 changed files with 1632 additions and 290 deletions
Download patch file Download diff file Expand all files Collapse all files

8
packages/pi-channels/src/adapters/slack.ts
View file

@ -39,8 +39,8 @@
import { SocketModeClient } from "@slack/socket-mode";
import { WebClient } from "@slack/web-api";
import { getChannelSetting } from "../config.ts";
import type { AdapterConfig, ChannelAdapter, ChannelMessage, OnIncomingMessage } from "../types.ts";
import { getChannelSetting } from "../config.js";
import type { AdapterConfig, ChannelAdapter, ChannelMessage, OnIncomingMessage } from "../types.js";
const MAX_LENGTH = 3000; // Slack block text limit; actual API limit is 4000 but leave margin
@ -146,7 +146,7 @@ export function createSlackAdapter(config: AdapterConfig, cwd?: string, log?: Sl
return {
direction: "bidirectional" as const,
async sendTyping(recipient: string): Promise<void> {
async sendTyping(_recipient: string): Promise<void> {
// Slack doesn't have a direct "typing" API for bots in channels.
// We can use a reaction or simply no-op. For DMs, there's no API either.
// Best we can do is nothing — Slack bots don't show typing indicators.
@ -309,7 +309,7 @@ export function createSlackAdapter(config: AdapterConfig, cwd?: string, log?: Sl
);
// ── Interactive payloads (future: button clicks, modals) ──
socketClient.on("interactive", async ({ body, ack }: { body: any; ack: () => Promise<void> }) => {
socketClient.on("interactive", async ({ body: _body, ack }: { body: any; ack: () => Promise<void> }) => {
try {
await ack();
// TODO: handle interactive payloads (block actions, modals)

11
packages/pi-channels/src/adapters/telegram.ts
View file

@ -36,8 +36,8 @@ import type {
IncomingMessage,
OnIncomingMessage,
TranscriptionConfig,
} from "../types.ts";
import { createTranscriptionProvider, type TranscriptionProvider } from "./transcription.ts";
} from "../types.js";
import { createTranscriptionProvider, type TranscriptionProvider } from "./transcription.js";
const MAX_LENGTH = 4096;
const MAX_FILE_SIZE = 1_048_576; // 1MB
@ -388,7 +388,6 @@ export function createTelegramAdapter(config: AdapterConfig): ChannelAdapter {
};
}
const ext = path.extname(filename || "").toLowerCase();
const attachment: IncomingAttachment = {
type: "image",
path: downloaded.localPath,
@ -472,7 +471,7 @@ export function createTelegramAdapter(config: AdapterConfig): ChannelAdapter {
return {
adapter: "telegram",
sender: chatId,
text: `🎵 ${filename || "audio"} (transcription failed${result.error ? ": " + result.error : ""})`,
text: `🎵 ${filename || "audio"} (transcription failed${result.error ? `: ${result.error}` : ""})`,
metadata: { ...metadata, hasAudio: true },
};
}
@ -535,7 +534,7 @@ export function createTelegramAdapter(config: AdapterConfig): ChannelAdapter {
return {
adapter: "telegram",
sender: chatId,
text: `🎤 (voice message — transcription failed${result.error ? ": " + result.error : ""})`,
text: `🎤 (voice message — transcription failed${result.error ? `: ${result.error}` : ""})`,
metadata: { ...metadata, hasVoice: true, voiceDuration: voice.duration },
};
}
@ -588,7 +587,7 @@ export function createTelegramAdapter(config: AdapterConfig): ChannelAdapter {
return {
adapter: "telegram",
sender: chatId,
text: `🎵 ${audioName} (transcription failed${result.error ? ": " + result.error : ""})`,
text: `🎵 ${audioName} (transcription failed${result.error ? `: ${result.error}` : ""})`,
metadata: { ...metadata, hasAudio: true, audioTitle: audio.title, audioDuration: audio.duration },
};
}

2
packages/pi-channels/src/adapters/transcription.ts
View file

@ -14,7 +14,7 @@
import { execFile } from "node:child_process";
import * as fs from "node:fs";
import * as path from "node:path";
import type { TranscriptionConfig } from "../types.ts";
import type { TranscriptionConfig } from "../types.js";
// ── Public interface ────────────────────────────────────────────

2
packages/pi-channels/src/adapters/webhook.ts
View file

@ -11,7 +11,7 @@
* }
*/
import type { AdapterConfig, ChannelAdapter, ChannelMessage } from "../types.ts";
import type { AdapterConfig, ChannelAdapter, ChannelMessage } from "../types.js";
export function createWebhookAdapter(config: AdapterConfig): ChannelAdapter {
const method = (config.method as string) ?? "POST";

202
packages/pi-channels/src/bridge/bridge.ts
View file

@ -2,18 +2,18 @@
* pi-channels — Chat bridge.
*
* Listens for incoming messages (channel:receive), serializes per sender,
* runs prompts via isolated subprocesses, and sends responses back via
* the same adapter. Each sender gets their own FIFO queue. Multiple
* senders run concurrently up to maxConcurrent.
* routes prompts into the live pi gateway runtime, and sends responses
* back via the same adapter. Each sender gets their own FIFO queue.
* Multiple senders run concurrently up to maxConcurrent.
*/
import type { EventBus } from "@mariozechner/pi-coding-agent";
import type { ChannelRegistry } from "../registry.ts";
import type { BridgeConfig, IncomingAttachment, IncomingMessage, QueuedPrompt, SenderSession } from "../types.ts";
import { type CommandContext, handleCommand, isCommand } from "./commands.ts";
import { RpcSessionManager } from "./rpc-runner.ts";
import { runPrompt } from "./runner.ts";
import { startTyping } from "./typing.ts";
import { readFileSync } from "node:fs";
import type { ImageContent } from "@mariozechner/pi-ai";
import { type EventBus, getActiveGatewayRuntime } from "@mariozechner/pi-coding-agent";
import type { ChannelRegistry } from "../registry.js";
import type { BridgeConfig, IncomingMessage, QueuedPrompt, SenderSession } from "../types.js";
import { type CommandContext, handleCommand, isCommand } from "./commands.js";
import { startTyping } from "./typing.js";
const BRIDGE_DEFAULTS: Required<BridgeConfig> = {
enabled: false,
@ -38,24 +38,21 @@ function nextId(): string {
export class ChatBridge {
private config: Required<BridgeConfig>;
private cwd: string;
private registry: ChannelRegistry;
private events: EventBus;
private log: LogFn;
private sessions = new Map<string, SenderSession>();
private activeCount = 0;
private running = false;
private rpcManager: RpcSessionManager | null = null;
constructor(
bridgeConfig: BridgeConfig | undefined,
cwd: string,
_cwd: string,
registry: ChannelRegistry,
events: EventBus,
log: LogFn = () => {},
) {
this.config = { ...BRIDGE_DEFAULTS, ...bridgeConfig };
this.cwd = cwd;
this.registry = registry;
this.events = events;
this.log = log;
@ -65,18 +62,11 @@ export class ChatBridge {
start(): void {
if (this.running) return;
if (!getActiveGatewayRuntime()) {
this.log("bridge-unavailable", { reason: "no active pi gateway runtime" }, "WARN");
return;
}
this.running = true;
// Always create the RPC manager — it's used on-demand for persistent senders
this.rpcManager = new RpcSessionManager(
{
cwd: this.cwd,
model: this.config.model,
timeoutMs: this.config.timeoutMs,
extensions: this.config.extensions,
},
this.config.idleTimeoutMinutes * 60_000,
);
}
stop(): void {
@ -86,8 +76,6 @@ export class ChatBridge {
}
this.sessions.clear();
this.activeCount = 0;
this.rpcManager?.killAll();
this.rpcManager = null;
}
isActive(): boolean {
@ -180,38 +168,32 @@ export class ChatBridge {
// Typing indicator
const adapter = this.registry.getAdapter(prompt.adapter);
const typing = this.config.typingIndicators ? startTyping(adapter, prompt.sender) : { stop() {} };
const ac = new AbortController();
session.abortController = ac;
const usePersistent = this.shouldUsePersistent(senderKey);
const gateway = getActiveGatewayRuntime();
if (!gateway) {
typing.stop();
session.processing = false;
this.activeCount--;
this.sendReply(prompt.adapter, prompt.sender, "❌ pi gateway is not running.");
return;
}
this.events.emit("bridge:start", {
id: prompt.id,
adapter: prompt.adapter,
sender: prompt.sender,
text: prompt.text.slice(0, 100),
persistent: usePersistent,
persistent: true,
});
try {
let result;
if (usePersistent && this.rpcManager) {
// Persistent mode: use RPC session
result = await this.runWithRpc(senderKey, prompt, ac.signal);
} else {
// Stateless mode: spawn subprocess
result = await runPrompt({
prompt: prompt.text,
cwd: this.cwd,
timeoutMs: this.config.timeoutMs,
model: this.config.model,
signal: ac.signal,
attachments: prompt.attachments,
extensions: this.config.extensions,
});
}
session.abortController = new AbortController();
const result = await gateway.enqueueMessage({
sessionKey: senderKey,
text: buildPromptText(prompt),
images: collectImageAttachments(prompt.attachments),
source: "extension",
metadata: prompt.metadata,
});
typing.stop();
@ -229,8 +211,7 @@ export class ChatBridge {
adapter: prompt.adapter,
sender: prompt.sender,
ok: result.ok,
durationMs: result.durationMs,
persistent: usePersistent,
persistent: true,
});
this.log(
"bridge-complete",
@ -238,15 +219,15 @@ export class ChatBridge {
id: prompt.id,
adapter: prompt.adapter,
ok: result.ok,
durationMs: result.durationMs,
persistent: usePersistent,
persistent: true,
},
result.ok ? "INFO" : "WARN",
);
} catch (err: any) {
} catch (err: unknown) {
typing.stop();
this.log("bridge-error", { adapter: prompt.adapter, sender: prompt.sender, error: err.message }, "ERROR");
this.sendReply(prompt.adapter, prompt.sender, `❌ Unexpected error: ${err.message}`);
const message = err instanceof Error ? err.message : String(err);
this.log("bridge-error", { adapter: prompt.adapter, sender: prompt.sender, error: message }, "ERROR");
this.sendReply(prompt.adapter, prompt.sender, `❌ Unexpected error: ${message}`);
} finally {
session.abortController = null;
session.processing = false;
@ -257,29 +238,6 @@ export class ChatBridge {
}
}
/** Run a prompt via persistent RPC session. */
private async runWithRpc(
senderKey: string,
prompt: QueuedPrompt,
signal?: AbortSignal,
): Promise<import("../types.ts").RunResult> {
try {
const rpcSession = await this.rpcManager!.getSession(senderKey);
return await rpcSession.runPrompt(prompt.text, {
signal,
attachments: prompt.attachments,
});
} catch (err: any) {
return {
ok: false,
response: "",
error: err.message,
durationMs: 0,
exitCode: 1,
};
}
}
/** After a slot frees up, check other senders waiting for concurrency. */
private drainWaiting(): void {
if (this.activeCount >= this.config.maxConcurrent) return;
@ -327,37 +285,17 @@ export class ChatBridge {
return this.sessions;
}
// ── Session mode resolution ───────────────────────────────
/**
* Determine if a sender should use persistent (RPC) or stateless mode.
* Checks sessionRules first (first match wins), falls back to sessionMode default.
*/
private shouldUsePersistent(senderKey: string): boolean {
for (const rule of this.config.sessionRules) {
if (globMatch(rule.match, senderKey)) {
return rule.mode === "persistent";
}
}
return this.config.sessionMode === "persistent";
}
// ── Command context ───────────────────────────────────────
private commandContext(): CommandContext {
const gateway = getActiveGatewayRuntime();
return {
isPersistent: (sender: string) => {
// Find the sender key to check mode
for (const [key, session] of this.sessions) {
if (session.sender === sender) return this.shouldUsePersistent(key);
}
return this.config.sessionMode === "persistent";
},
isPersistent: () => true,
abortCurrent: (sender: string): boolean => {
for (const session of this.sessions.values()) {
if (!gateway) return false;
for (const [key, session] of this.sessions) {
if (session.sender === sender && session.abortController) {
session.abortController.abort();
return true;
return gateway.abortSession(key);
}
}
return false;
@ -368,13 +306,11 @@ export class ChatBridge {
}
},
resetSession: (sender: string): void => {
if (!gateway) return;
for (const [key, session] of this.sessions) {
if (session.sender === sender) {
this.sessions.delete(key);
// Also reset persistent RPC session
if (this.rpcManager) {
this.rpcManager.resetSession(key).catch(() => {});
}
void gateway.resetSession(key);
}
}
},
@ -388,21 +324,6 @@ export class ChatBridge {
}
}
// ── Helpers ───────────────────────────────────────────────────
/**
* Simple glob matcher supporting `*` (any chars) and `?` (single char).
* Used for sessionRules pattern matching against "adapter:senderId" keys.
*/
function globMatch(pattern: string, text: string): boolean {
// Escape regex special chars except * and ?
const re = pattern
.replace(/[.+^${}()|[\]\\]/g, "\\$&")
.replace(/\*/g, ".*")
.replace(/\?/g, ".");
return new RegExp(`^${re}$`).test(text);
}
const MAX_ERROR_LENGTH = 200;
/**
@ -428,5 +349,36 @@ function sanitizeError(error: string | undefined): string {
const msg = meaningful?.trim() || "Something went wrong. Please try again.";
return msg.length > MAX_ERROR_LENGTH ? msg.slice(0, MAX_ERROR_LENGTH) + "…" : msg;
return msg.length > MAX_ERROR_LENGTH ? `${msg.slice(0, MAX_ERROR_LENGTH)}…` : msg;
}
function collectImageAttachments(attachments: QueuedPrompt["attachments"]): ImageContent[] | undefined {
if (!attachments || attachments.length === 0) {
return undefined;
}
const images = attachments
.filter((attachment) => attachment.type === "image")
.map((attachment) => ({
type: "image" as const,
data: readFileSync(attachment.path).toString("base64"),
mimeType: attachment.mimeType || "image/jpeg",
}));
return images.length > 0 ? images : undefined;
}
function buildPromptText(prompt: QueuedPrompt): string {
if (!prompt.attachments || prompt.attachments.length === 0) {
return prompt.text;
}
const attachmentNotes = prompt.attachments
.filter((attachment) => attachment.type !== "image")
.map((attachment) => {
const label = attachment.filename ?? attachment.path;
return `Attachment (${attachment.type}): ${label}`;
});
if (attachmentNotes.length === 0) {
return prompt.text;
}
return `${prompt.text}\n\n${attachmentNotes.join("\n")}`;
}

2
packages/pi-channels/src/bridge/commands.ts
View file

@ -7,7 +7,7 @@
* Built-in: /start, /help, /abort, /status, /new
*/
import type { SenderSession } from "../types.ts";
import type { SenderSession } from "../types.js";
export interface BotCommand {
name: string;

182
packages/pi-channels/src/bridge/rpc-runner.ts
View file

@ -14,7 +14,7 @@
import { type ChildProcess, spawn } from "node:child_process";
import * as readline from "node:readline";
import type { IncomingAttachment, RunResult } from "../types.ts";
import type { IncomingAttachment, RunResult } from "../types.js";
export interface RpcRunnerOptions {
cwd: string;
@ -118,95 +118,97 @@ export class RpcSession {
onStreaming?: (text: string) => void;
},
): Promise<RunResult> {
return new Promise(async (resolve) => {
// Ensure subprocess is running
if (!this.ready) {
const ok = await this.start();
if (!ok) {
resolve({
ok: false,
response: "",
error: "Failed to start RPC session",
durationMs: 0,
exitCode: 1,
});
return;
}
}
const startTime = Date.now();
this._onStreaming = options?.onStreaming ?? null;
// Timeout
const timer = setTimeout(() => {
if (this.pending) {
const p = this.pending;
this.pending = null;
const text = p.textChunks.join("");
p.resolve({
ok: false,
response: text || "(timed out)",
error: "Timeout",
durationMs: Date.now() - p.startTime,
exitCode: 124,
});
// Kill and restart on next message
this.cleanup();
}
}, this.options.timeoutMs);
this.pending = { resolve, startTime, timer, textChunks: [] };
// Abort handler
const onAbort = () => {
this.sendCommand({ type: "abort" });
};
if (options?.signal) {
if (options.signal.aborted) {
clearTimeout(timer);
this.pending = null;
this.sendCommand({ type: "abort" });
resolve({
ok: false,
response: "(aborted)",
error: "Aborted by user",
durationMs: Date.now() - startTime,
exitCode: 130,
});
return;
}
options.signal.addEventListener("abort", onAbort, { once: true });
this.pending.abortHandler = () => options.signal?.removeEventListener("abort", onAbort);
}
// Build prompt command
const cmd: Record<string, unknown> = {
type: "prompt",
message: prompt,
};
// Attach images as base64
if (options?.attachments?.length) {
const images: Array<Record<string, string>> = [];
for (const att of options.attachments) {
if (att.type === "image") {
try {
const fs = await import("node:fs");
const data = fs.readFileSync(att.path).toString("base64");
images.push({
type: "image",
data,
mimeType: att.mimeType || "image/jpeg",
});
} catch {
// Skip unreadable attachments
}
return new Promise((resolve) => {
void (async () => {
// Ensure subprocess is running
if (!this.ready) {
const ok = await this.start();
if (!ok) {
resolve({
ok: false,
response: "",
error: "Failed to start RPC session",
durationMs: 0,
exitCode: 1,
});
return;
}
}
if (images.length > 0) cmd.images = images;
}
this.sendCommand(cmd);
const startTime = Date.now();
this._onStreaming = options?.onStreaming ?? null;
// Timeout
const timer = setTimeout(() => {
if (this.pending) {
const p = this.pending;
this.pending = null;
const text = p.textChunks.join("");
p.resolve({
ok: false,
response: text || "(timed out)",
error: "Timeout",
durationMs: Date.now() - p.startTime,
exitCode: 124,
});
// Kill and restart on next message
this.cleanup();
}
}, this.options.timeoutMs);
this.pending = { resolve, startTime, timer, textChunks: [] };
// Abort handler
const onAbort = () => {
this.sendCommand({ type: "abort" });
};
if (options?.signal) {
if (options.signal.aborted) {
clearTimeout(timer);
this.pending = null;
this.sendCommand({ type: "abort" });
resolve({
ok: false,
response: "(aborted)",
error: "Aborted by user",
durationMs: Date.now() - startTime,
exitCode: 130,
});
return;
}
options.signal.addEventListener("abort", onAbort, { once: true });
this.pending.abortHandler = () => options.signal?.removeEventListener("abort", onAbort);
}
// Build prompt command
const cmd: Record<string, unknown> = {
type: "prompt",
message: prompt,
};
// Attach images as base64
if (options?.attachments?.length) {
const images: Array<Record<string, string>> = [];
for (const att of options.attachments) {
if (att.type === "image") {
try {
const fs = await import("node:fs");
const data = fs.readFileSync(att.path).toString("base64");
images.push({
type: "image",
data,
mimeType: att.mimeType || "image/jpeg",
});
} catch {
// Skip unreadable attachments
}
}
}
if (images.length > 0) cmd.images = images;
}
this.sendCommand(cmd);
})();
});
}
@ -253,7 +255,7 @@ export class RpcSession {
private sendCommand(cmd: Record<string, unknown>): void {
if (!this.child?.stdin?.writable) return;
this.child.stdin.write(JSON.stringify(cmd) + "\n");
this.child.stdin.write(`${JSON.stringify(cmd)}\n`);
}
private handleLine(line: string): void {
@ -358,7 +360,7 @@ export class RpcSessionManager {
/** Get or create a session for a sender. */
async getSession(senderKey: string): Promise<RpcSession> {
let session = this.sessions.get(senderKey);
if (session && session.isAlive()) {
if (session?.isAlive()) {
this.resetIdleTimer(senderKey);
return session;
}
@ -403,7 +405,7 @@ export class RpcSessionManager {
/** Kill all sessions. */
killAll(): void {
for (const [key, session] of this.sessions) {
for (const session of this.sessions.values()) {
session.cleanup();
}
this.sessions.clear();

2
packages/pi-channels/src/bridge/runner.ts
View file

@ -7,7 +7,7 @@
*/
import { type ChildProcess, spawn } from "node:child_process";
import type { IncomingAttachment, RunResult } from "../types.ts";
import type { IncomingAttachment, RunResult } from "../types.js";
export interface RunOptions {
prompt: string;

2
packages/pi-channels/src/bridge/typing.ts
View file

@ -6,7 +6,7 @@
* For adapters without sendTyping, this is a no-op.
*/
import type { ChannelAdapter } from "../types.ts";
import type { ChannelAdapter } from "../types.js";
const TYPING_INTERVAL_MS = 4_000;

2
packages/pi-channels/src/config.ts
View file

@ -29,7 +29,7 @@
*/
import { getAgentDir, SettingsManager } from "@mariozechner/pi-coding-agent";
import type { ChannelConfig } from "./types.ts";
import type { ChannelConfig } from "./types.js";
const SETTINGS_KEY = "pi-channels";

6
packages/pi-channels/src/events.ts
View file

@ -15,9 +15,9 @@
*/
import type { ExtensionAPI } from "@mariozechner/pi-coding-agent";
import type { ChatBridge } from "./bridge/bridge.ts";
import type { ChannelRegistry } from "./registry.ts";
import type { ChannelAdapter, ChannelMessage, IncomingMessage } from "./types.ts";
import type { ChatBridge } from "./bridge/bridge.js";
import type { ChannelRegistry } from "./registry.js";
import type { ChannelAdapter, ChannelMessage, IncomingMessage } from "./types.js";
/** Reference to the active bridge, set by index.ts after construction. */
let activeBridge: ChatBridge | null = null;

12
packages/pi-channels/src/index.ts
View file

@ -35,12 +35,12 @@
*/
import type { ExtensionAPI } from "@mariozechner/pi-coding-agent";
import { ChatBridge } from "./bridge/bridge.ts";
import { loadConfig } from "./config.ts";
import { registerChannelEvents, setBridge } from "./events.ts";
import { createLogger } from "./logger.ts";
import { ChannelRegistry } from "./registry.ts";
import { registerChannelTool } from "./tool.ts";
import { ChatBridge } from "./bridge/bridge.js";
import { loadConfig } from "./config.js";
import { registerChannelEvents, setBridge } from "./events.js";
import { createLogger } from "./logger.js";
import { ChannelRegistry } from "./registry.js";
import { registerChannelTool } from "./tool.js";
export default function (pi: ExtensionAPI) {
const log = createLogger(pi);

8
packages/pi-channels/src/registry.ts
View file

@ -2,9 +2,9 @@
* pi-channels — Adapter registry + route resolution.
*/
import { createSlackAdapter } from "./adapters/slack.ts";
import { createTelegramAdapter } from "./adapters/telegram.ts";
import { createWebhookAdapter } from "./adapters/webhook.ts";
import { createSlackAdapter } from "./adapters/slack.js";
import { createTelegramAdapter } from "./adapters/telegram.js";
import { createWebhookAdapter } from "./adapters/webhook.js";
import type {
AdapterConfig,
AdapterDirection,
@ -13,7 +13,7 @@ import type {
ChannelMessage,
IncomingMessage,
OnIncomingMessage,
} from "./types.ts";
} from "./types.js";
// ── Built-in adapter factories ──────────────────────────────────

2
packages/pi-channels/src/tool.ts
View file

@ -5,7 +5,7 @@
import { StringEnum } from "@mariozechner/pi-ai";
import type { ExtensionAPI } from "@mariozechner/pi-coding-agent";
import { Type } from "@sinclair/typebox";
import type { ChannelRegistry } from "./registry.ts";
import type { ChannelRegistry } from "./registry.js";
interface ChannelToolParams {
action: "send" | "list" | "test";