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

fix(runtime): keep daemon alive and localize package installs

Browse source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Harivansh Rathi 2026-03-05 17:36:25 -08:00
parent fa208bca73
commit 3f04822f58
38 changed files with 2051 additions and 1939 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -39,13 +39,8 @@
import { SocketModeClient } from "@slack/socket-mode";
import { WebClient } from "@slack/web-api";
import type {
ChannelAdapter,
ChannelMessage,
AdapterConfig,
OnIncomingMessage,
} from "../types.ts";
import { getChannelSetting } from "../config.ts";
import type { AdapterConfig, ChannelAdapter, ChannelMessage, OnIncomingMessage } from "../types.ts";
const MAX_LENGTH = 3000; // Slack block text limit; actual API limit is 4000 but leave margin
@ -88,17 +83,17 @@ export type SlackAdapterLogger = (event: string, data: Record<string, unknown>,
export function createSlackAdapter(config: AdapterConfig, cwd?: string, log?: SlackAdapterLogger): ChannelAdapter {
// Tokens live in settings under pi-channels.slack (not in the adapter config block)
const appToken = (cwd ? getChannelSetting(cwd, "slack.appToken") as string : null)
?? config.appToken as string;
const botToken = (cwd ? getChannelSetting(cwd, "slack.botToken") as string : null)
?? config.botToken as string;
const appToken = (cwd ? (getChannelSetting(cwd, "slack.appToken") as string) : null) ?? (config.appToken as string);
const botToken = (cwd ? (getChannelSetting(cwd, "slack.botToken") as string) : null) ?? (config.botToken as string);
const allowedChannelIds = config.allowedChannelIds as string[] | undefined;
const respondToMentionsOnly = config.respondToMentionsOnly === true;
const slashCommand = (config.slashCommand as string) ?? "/aivena";
if (!appToken) throw new Error("Slack adapter requires appToken (xapp-...) in settings under pi-channels.slack.appToken");
if (!botToken) throw new Error("Slack adapter requires botToken (xoxb-...) in settings under pi-channels.slack.botToken");
if (!appToken)
throw new Error("Slack adapter requires appToken (xapp-...) in settings under pi-channels.slack.appToken");
if (!botToken)
throw new Error("Slack adapter requires botToken (xoxb-...) in settings under pi-channels.slack.botToken");
let socketClient: SocketModeClient | null = null;
const webClient = new WebClient(botToken);
@ -119,7 +114,10 @@ export function createSlackAdapter(config: AdapterConfig, cwd?: string, log?: Sl
}
/** Build metadata common to all incoming messages */
function buildMetadata(event: { channel?: string; user?: string; ts?: string; thread_ts?: string; channel_type?: string }, extra?: Record<string, unknown>): Record<string, unknown> {
function buildMetadata(
event: { channel?: string; user?: string; ts?: string; thread_ts?: string; channel_type?: string },
extra?: Record<string, unknown>,
): Record<string, unknown> {
return {
channelId: event.channel,
userId: event.user,
@ -184,7 +182,7 @@ export function createSlackAdapter(config: AdapterConfig, cwd?: string, log?: Sl
// Resolve bot user ID (for stripping self-mentions)
try {
const authResult = await webClient.auth.test();
botUserId = authResult.user_id as string ?? null;
botUserId = (authResult.user_id as string) ?? null;
} catch {
// Non-fatal — mention stripping just won't work
}
@ -215,16 +213,20 @@ export function createSlackAdapter(config: AdapterConfig, cwd?: string, log?: Sl
// handled by the app_mention listener to avoid duplicate responses.
// DMs (im) and multi-party DMs (mpim) don't fire app_mention, so we
// must NOT skip those here.
if (botUserId && (event.channel_type === "channel" || event.channel_type === "group") && event.text.includes(`<@${botUserId}>`)) return;
if (
botUserId &&
(event.channel_type === "channel" || event.channel_type === "group") &&
event.text.includes(`<@${botUserId}>`)
)
return;
// In channels/groups, optionally only respond to @mentions
// (app_mention events are handled separately below)
if (respondToMentionsOnly && (event.channel_type === "channel" || event.channel_type === "group")) return;
if (respondToMentionsOnly && (event.channel_type === "channel" || event.channel_type === "group"))
return;
// Use channel:threadTs as sender key for threaded conversations
const sender = event.thread_ts
? `${event.channel}:${event.thread_ts}`
: event.channel;
const sender = event.thread_ts ? `${event.channel}:${event.thread_ts}` : event.channel;
onMessage({
adapter: "slack",
@ -234,74 +236,86 @@ export function createSlackAdapter(config: AdapterConfig, cwd?: string, log?: Sl
eventType: "message",
}),
});
} catch (err) { log?.("slack-handler-error", { handler: "message", error: String(err) }, "ERROR"); }
} catch (err) {
log?.("slack-handler-error", { handler: "message", error: String(err) }, "ERROR");
}
});
// ── App mention events ──────────────────────────
socketClient.on("app_mention", async ({ event, ack }: { event: SlackMentionEvent; ack: () => Promise<void> }) => {
try {
await ack();
socketClient.on(
"app_mention",
async ({ event, ack }: { event: SlackMentionEvent; ack: () => Promise<void> }) => {
try {
await ack();
if (!isAllowed(event.channel)) return;
if (!isAllowed(event.channel)) return;
const sender = event.thread_ts
? `${event.channel}:${event.thread_ts}`
: event.channel;
const sender = event.thread_ts ? `${event.channel}:${event.thread_ts}` : event.channel;
onMessage({
adapter: "slack",
sender,
text: stripBotMention(event.text),
metadata: buildMetadata(event, {
eventType: "app_mention",
}),
});
} catch (err) { log?.("slack-handler-error", { handler: "app_mention", error: String(err) }, "ERROR"); }
});
onMessage({
adapter: "slack",
sender,
text: stripBotMention(event.text),
metadata: buildMetadata(event, {
eventType: "app_mention",
}),
});
} catch (err) {
log?.("slack-handler-error", { handler: "app_mention", error: String(err) }, "ERROR");
}
},
);
// ── Slash commands ───────────────────────────────
socketClient.on("slash_commands", async ({ body, ack }: { body: SlackCommandPayload; ack: (response?: any) => Promise<void> }) => {
try {
if (body.command !== slashCommand) {
await ack();
return;
socketClient.on(
"slash_commands",
async ({ body, ack }: { body: SlackCommandPayload; ack: (response?: any) => Promise<void> }) => {
try {
if (body.command !== slashCommand) {
await ack();
return;
}
if (!body.text?.trim()) {
await ack({ text: `Usage: ${slashCommand} [your message]` });
return;
}
if (!isAllowed(body.channel_id)) {
await ack({ text: "⛔ This command is not available in this channel." });
return;
}
// Acknowledge immediately (Slack requires <3s response)
await ack({ text: "🤔 Thinking..." });
onMessage({
adapter: "slack",
sender: body.channel_id,
text: body.text.trim(),
metadata: {
channelId: body.channel_id,
channelName: body.channel_name,
userId: body.user_id,
userName: body.user_name,
eventType: "slash_command",
command: body.command,
},
});
} catch (err) {
log?.("slack-handler-error", { handler: "slash_commands", error: String(err) }, "ERROR");
}
if (!body.text?.trim()) {
await ack({ text: `Usage: ${slashCommand} [your message]` });
return;
}
if (!isAllowed(body.channel_id)) {
await ack({ text: "⛔ This command is not available in this channel." });
return;
}
// Acknowledge immediately (Slack requires <3s response)
await ack({ text: "🤔 Thinking..." });
onMessage({
adapter: "slack",
sender: body.channel_id,
text: body.text.trim(),
metadata: {
channelId: body.channel_id,
channelName: body.channel_name,
userId: body.user_id,
userName: body.user_name,
eventType: "slash_command",
command: body.command,
},
});
} catch (err) { log?.("slack-handler-error", { handler: "slash_commands", error: String(err) }, "ERROR"); }
});
},
);
// ── Interactive payloads (future: button clicks, modals) ──
socketClient.on("interactive", async ({ body, ack }: { body: any; ack: () => Promise<void> }) => {
try {
await ack();
// TODO: handle interactive payloads (block actions, modals)
} catch (err) { log?.("slack-handler-error", { handler: "interactive", error: String(err) }, "ERROR"); }
} catch (err) {
log?.("slack-handler-error", { handler: "interactive", error: String(err) }, "ERROR");
}
});
await socketClient.start();

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

@ -26,15 +26,15 @@
*/
import * as fs from "node:fs";
import * as path from "node:path";
import * as os from "node:os";
import * as path from "node:path";
import type {
AdapterConfig,
ChannelAdapter,
ChannelMessage,
AdapterConfig,
OnIncomingMessage,
IncomingMessage,
IncomingAttachment,
IncomingMessage,
OnIncomingMessage,
TranscriptionConfig,
} from "../types.ts";
import { createTranscriptionProvider, type TranscriptionProvider } from "./transcription.ts";
@ -63,12 +63,49 @@ const TEXT_MIME_TYPES = new Set([
/** File extensions we treat as text even if MIME is generic (application/octet-stream). */
const TEXT_EXTENSIONS = new Set([
".md", ".markdown", ".txt", ".csv", ".json", ".jsonl", ".yaml", ".yml",
".toml", ".xml", ".html", ".htm", ".css", ".js", ".ts", ".tsx", ".jsx",
".py", ".rs", ".go", ".rb", ".php", ".java", ".kt", ".c", ".cpp", ".h",
".sh", ".bash", ".zsh", ".fish", ".sql", ".graphql", ".gql",
".env", ".ini", ".cfg", ".conf", ".properties", ".log",
".gitignore", ".dockerignore", ".editorconfig",
".md",
".markdown",
".txt",
".csv",
".json",
".jsonl",
".yaml",
".yml",
".toml",
".xml",
".html",
".htm",
".css",
".js",
".ts",
".tsx",
".jsx",
".py",
".rs",
".go",
".rb",
".php",
".java",
".kt",
".c",
".cpp",
".h",
".sh",
".bash",
".zsh",
".fish",
".sql",
".graphql",
".gql",
".env",
".ini",
".cfg",
".conf",
".properties",
".log",
".gitignore",
".dockerignore",
".editorconfig",
]);
/** Image MIME prefixes. */
@ -80,15 +117,22 @@ function isImageMime(mime: string | undefined): boolean {
/** Audio MIME types that can be transcribed. */
const AUDIO_MIME_PREFIXES = ["audio/"];
const AUDIO_MIME_TYPES = new Set([
"audio/mpeg", "audio/mp4", "audio/ogg", "audio/wav", "audio/webm",
"audio/x-m4a", "audio/flac", "audio/aac", "audio/mp3",
"audio/mpeg",
"audio/mp4",
"audio/ogg",
"audio/wav",
"audio/webm",
"audio/x-m4a",
"audio/flac",
"audio/aac",
"audio/mp3",
"video/ogg", // .ogg containers can be audio-only
]);
function isAudioMime(mime: string | undefined): boolean {
if (!mime) return false;
if (AUDIO_MIME_TYPES.has(mime)) return true;
return AUDIO_MIME_PREFIXES.some(p => mime.startsWith(p));
return AUDIO_MIME_PREFIXES.some((p) => mime.startsWith(p));
}
function isTextDocument(mimeType: string | undefined, filename: string | undefined): boolean {
@ -166,13 +210,17 @@ export function createTelegramAdapter(config: AdapterConfig): ChannelAdapter {
* Download a file from Telegram by file_id.
* Returns { path, size } or null on failure.
*/
async function downloadFile(fileId: string, suggestedName?: string, maxSize = MAX_FILE_SIZE): Promise<{ localPath: string; size: number } | null> {
async function downloadFile(
fileId: string,
suggestedName?: string,
maxSize = MAX_FILE_SIZE,
): Promise<{ localPath: string; size: number } | null> {
try {
// Get file info
const infoRes = await fetch(`${apiBase}/getFile?file_id=${fileId}`);
if (!infoRes.ok) return null;
const info = await infoRes.json() as {
const info = (await infoRes.json()) as {
ok: boolean;
result?: { file_id: string; file_size?: number; file_path?: string };
};
@ -237,7 +285,7 @@ export function createTelegramAdapter(config: AdapterConfig): ChannelAdapter {
continue;
}
const data = await res.json() as {
const data = (await res.json()) as {
ok: boolean;
result: Array<{ update_id: number; message?: TelegramMessage }>;
};
@ -574,7 +622,11 @@ export function createTelegramAdapter(config: AdapterConfig): ChannelAdapter {
function cleanupTempFiles(): void {
for (const f of tempFiles) {
try { fs.unlinkSync(f); } catch { /* ignore */ }
try {
fs.unlinkSync(f);
} catch {
/* ignore */
}
}
tempFiles.length = 0;
}
@ -663,7 +715,7 @@ interface TelegramMessage {
}
function sleep(ms: number): Promise<void> {
return new Promise(resolve => setTimeout(resolve, ms));
return new Promise((resolve) => setTimeout(resolve, ms));
}
function formatSize(bytes: number): string {

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

@ -11,9 +11,9 @@
* const result = await provider.transcribe("/path/to/audio.ogg", "en");
*/
import { execFile } from "node:child_process";
import * as fs from "node:fs";
import * as path from "node:path";
import { execFile } from "node:child_process";
import type { TranscriptionConfig } from "../types.ts";
// ── Public interface ────────────────────────────────────────────
@ -187,7 +187,7 @@ class OpenAIProvider implements TranscriptionProvider {
return { ok: false, error: `OpenAI API error (${response.status}): ${body.slice(0, 200)}` };
}
const data = await response.json() as { text?: string };
const data = (await response.json()) as { text?: string };
if (!data.text) {
return { ok: false, error: "OpenAI returned empty transcription" };
}
@ -238,7 +238,7 @@ class ElevenLabsProvider implements TranscriptionProvider {
return { ok: false, error: `ElevenLabs API error (${response.status}): ${body.slice(0, 200)}` };
}
const data = await response.json() as { text?: string };
const data = (await response.json()) as { text?: string };
if (!data.text) {
return { ok: false, error: "ElevenLabs returned empty transcription" };
}

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

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

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

@ -7,18 +7,12 @@
* senders run concurrently up to maxConcurrent.
*/
import type {
IncomingMessage,
IncomingAttachment,
QueuedPrompt,
SenderSession,
BridgeConfig,
} from "../types.ts";
import type { ChannelRegistry } from "../registry.ts";
import type { EventBus } from "@mariozechner/pi-coding-agent";
import { runPrompt } from "./runner.ts";
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 { isCommand, handleCommand, type CommandContext } from "./commands.ts";
import { runPrompt } from "./runner.ts";
import { startTyping } from "./typing.ts";
const BRIDGE_DEFAULTS: Required<BridgeConfig> = {
@ -144,7 +138,7 @@ export class ChatBridge {
message.adapter,
message.sender,
`⚠️ Queue full (${this.config.maxQueuePerSender} pending). ` +
`Wait for current prompts to finish or use /abort.`,
`Wait for current prompts to finish or use /abort.`,
);
return;
}
@ -163,7 +157,9 @@ export class ChatBridge {
session.messageCount++;
this.events.emit("bridge:enqueue", {
id: queued.id, adapter: message.adapter, sender: message.sender,
id: queued.id,
adapter: message.adapter,
sender: message.sender,
queueDepth: session.queue.length,
});
@ -183,9 +179,7 @@ export class ChatBridge {
// Typing indicator
const adapter = this.registry.getAdapter(prompt.adapter);
const typing = this.config.typingIndicators
? startTyping(adapter, prompt.sender)
: { stop() {} };
const typing = this.config.typingIndicators ? startTyping(adapter, prompt.sender) : { stop() {} };
const ac = new AbortController();
session.abortController = ac;
@ -193,7 +187,9 @@ export class ChatBridge {
const usePersistent = this.shouldUsePersistent(senderKey);
this.events.emit("bridge:start", {
id: prompt.id, adapter: prompt.adapter, sender: prompt.sender,
id: prompt.id,
adapter: prompt.adapter,
sender: prompt.sender,
text: prompt.text.slice(0, 100),
persistent: usePersistent,
});
@ -225,22 +221,28 @@ export class ChatBridge {
this.sendReply(prompt.adapter, prompt.sender, "⏹ Aborted.");
} else {
const userError = sanitizeError(result.error);
this.sendReply(
prompt.adapter, prompt.sender,
result.response || `❌ ${userError}`,
);
this.sendReply(prompt.adapter, prompt.sender, result.response || `❌ ${userError}`);
}
this.events.emit("bridge:complete", {
id: prompt.id, adapter: prompt.adapter, sender: prompt.sender,
ok: result.ok, durationMs: result.durationMs,
id: prompt.id,
adapter: prompt.adapter,
sender: prompt.sender,
ok: result.ok,
durationMs: result.durationMs,
persistent: usePersistent,
});
this.log("bridge-complete", {
id: prompt.id, adapter: prompt.adapter, ok: result.ok,
durationMs: result.durationMs, persistent: usePersistent,
}, result.ok ? "INFO" : "WARN");
this.log(
"bridge-complete",
{
id: prompt.id,
adapter: prompt.adapter,
ok: result.ok,
durationMs: result.durationMs,
persistent: usePersistent,
},
result.ok ? "INFO" : "WARN",
);
} catch (err: any) {
typing.stop();
this.log("bridge-error", { adapter: prompt.adapter, sender: prompt.sender, error: err.message }, "ERROR");
@ -296,9 +298,7 @@ export class ChatBridge {
adapter: message.adapter,
sender: message.sender,
displayName:
(message.metadata?.firstName as string) ||
(message.metadata?.username as string) ||
message.sender,
(message.metadata?.firstName as string) || (message.metadata?.username as string) || message.sender,
queue: [],
processing: false,
abortController: null,
@ -413,21 +413,20 @@ function sanitizeError(error: string | undefined): string {
if (!error) return "Something went wrong. Please try again.";
// Extract the most meaningful line — skip "Extension error" noise and stack traces
const lines = error.split("\n").filter(l => l.trim());
const lines = error.split("\n").filter((l) => l.trim());
// Find the first line that isn't an extension loading error or stack frame
const meaningful = lines.find(l =>
!l.startsWith("Extension error") &&
!l.startsWith(" at ") &&
!l.startsWith("node:") &&
!l.includes("NODE_MODULE_VERSION") &&
!l.includes("compiled against a different") &&
!l.includes("Emitted 'error' event")
const meaningful = lines.find(
(l) =>
!l.startsWith("Extension error") &&
!l.startsWith(" at ") &&
!l.startsWith("node:") &&
!l.includes("NODE_MODULE_VERSION") &&
!l.includes("compiled against a different") &&
!l.includes("Emitted 'error' event"),
);
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;
}

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

@ -51,11 +51,7 @@ export function getAllCommands(): BotCommand[] {
* Handle a command. Returns reply text, or null if unrecognized
* (fall through to agent).
*/
export function handleCommand(
text: string,
session: SenderSession | undefined,
ctx: CommandContext,
): string | null {
export function handleCommand(text: string, session: SenderSession | undefined, ctx: CommandContext): string | null {
const { command } = parseCommand(text);
if (!command) return null;
const cmd = commands.get(command);
@ -89,9 +85,7 @@ registerCommand({
handler: (_args, session, ctx) => {
if (!session) return "No active session.";
if (!session.processing) return "Nothing is running right now.";
return ctx.abortCurrent(session.sender)
? "⏹ Aborting current prompt..."
: "Failed to abort — nothing running.";
return ctx.abortCurrent(session.sender) ? "⏹ Aborting current prompt..." : "Failed to abort — nothing running.";
},
});

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

@ -12,9 +12,9 @@
* 4. Subprocess crash triggers auto-restart on next message
*/
import { spawn, type ChildProcess } from "node:child_process";
import { type ChildProcess, spawn } from "node:child_process";
import * as readline from "node:readline";
import type { RunResult, IncomingAttachment } from "../types.ts";
import type { IncomingAttachment, RunResult } from "../types.ts";
export interface RpcRunnerOptions {
cwd: string;
@ -176,8 +176,7 @@ export class RpcSession {
return;
}
options.signal.addEventListener("abort", onAbort, { once: true });
this.pending.abortHandler = () =>
options.signal?.removeEventListener("abort", onAbort);
this.pending.abortHandler = () => options.signal?.removeEventListener("abort", onAbort);
}
// Build prompt command

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

@ -6,8 +6,8 @@
* Same pattern as pi-cron and pi-heartbeat.
*/
import { spawn, type ChildProcess } from "node:child_process";
import type { RunResult, IncomingAttachment } from "../types.ts";
import { type ChildProcess, spawn } from "node:child_process";
import type { IncomingAttachment, RunResult } from "../types.ts";
export interface RunOptions {
prompt: string;
@ -56,25 +56,37 @@ export function runPrompt(options: RunOptions): Promise<RunResult> {
});
} catch (err: any) {
resolve({
ok: false, response: "", error: `Failed to spawn: ${err.message}`,
durationMs: Date.now() - startTime, exitCode: 1,
ok: false,
response: "",
error: `Failed to spawn: ${err.message}`,
durationMs: Date.now() - startTime,
exitCode: 1,
});
return;
}
let stdout = "";
let stderr = "";
child.stdout?.on("data", (chunk: Buffer) => { stdout += chunk.toString(); });
child.stderr?.on("data", (chunk: Buffer) => { stderr += chunk.toString(); });
child.stdout?.on("data", (chunk: Buffer) => {
stdout += chunk.toString();
});
child.stderr?.on("data", (chunk: Buffer) => {
stderr += chunk.toString();
});
const onAbort = () => {
child.kill("SIGTERM");
setTimeout(() => { if (!child.killed) child.kill("SIGKILL"); }, 3000);
setTimeout(() => {
if (!child.killed) child.kill("SIGKILL");
}, 3000);
};
if (signal) {
if (signal.aborted) { onAbort(); }
else { signal.addEventListener("abort", onAbort, { once: true }); }
if (signal.aborted) {
onAbort();
} else {
signal.addEventListener("abort", onAbort, { once: true });
}
}
child.on("close", (code) => {
@ -84,7 +96,13 @@ export function runPrompt(options: RunOptions): Promise<RunResult> {
const exitCode = code ?? 1;
if (signal?.aborted) {
resolve({ ok: false, response: response || "(aborted)", error: "Aborted by user", durationMs, exitCode: 130 });
resolve({
ok: false,
response: response || "(aborted)",
error: "Aborted by user",
durationMs,
exitCode: 130,
});
} else if (exitCode !== 0) {
resolve({ ok: false, response, error: stderr.trim() || `Exit code ${exitCode}`, durationMs, exitCode });
} else {

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

@ -14,10 +14,7 @@ const TYPING_INTERVAL_MS = 4_000;
* Start sending typing indicators. Returns a stop() handle.
* No-op if the adapter doesn't support sendTyping.
*/
export function startTyping(
adapter: ChannelAdapter | undefined,
recipient: string,
): { stop: () => void } {
export function startTyping(adapter: ChannelAdapter | undefined, recipient: string): { stop: () => void } {
if (!adapter?.sendTyping) return { stop() {} };
// Fire immediately

29
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.ts";
/** Reference to the active bridge, set by index.ts after construction. */
let activeBridge: ChatBridge | null = null;
@ -27,7 +27,6 @@ export function setBridge(bridge: ChatBridge | null): void {
}
export function registerChannelEvents(pi: ExtensionAPI, registry: ChannelRegistry): void {
// ── Incoming messages → channel:receive (+ bridge) ──────
registry.setOnIncoming((message: IncomingMessage) => {
@ -53,9 +52,7 @@ export function registerChannelEvents(pi: ExtensionAPI, registry: ChannelRegistr
if (!event.job.channel) return;
if (!event.response && !event.error) return;
const text = event.ok
? event.response ?? "(no output)"
: `❌ Error: ${event.error ?? "unknown"}`;
const text = event.ok ? (event.response ?? "(no output)") : `❌ Error: ${event.error ?? "unknown"}`;
registry.send({
adapter: event.job.channel,
@ -70,7 +67,7 @@ export function registerChannelEvents(pi: ExtensionAPI, registry: ChannelRegistr
pi.events.on("channel:send", (raw: unknown) => {
const data = raw as ChannelMessage & { callback?: (result: { ok: boolean; error?: string }) => void };
registry.send(data).then(r => data.callback?.(r));
registry.send(data).then((r) => data.callback?.(r));
});
// ── channel:register — add a custom adapter ──────────────
@ -102,12 +99,18 @@ export function registerChannelEvents(pi: ExtensionAPI, registry: ChannelRegistr
// ── channel:test — send a test ping ──────────────────────
pi.events.on("channel:test", (raw: unknown) => {
const data = raw as { adapter: string; recipient: string; callback?: (result: { ok: boolean; error?: string }) => void };
registry.send({
adapter: data.adapter,
recipient: data.recipient ?? "",
text: `🏓 pi-channels test — ${new Date().toISOString()}`,
source: "channel:test",
}).then(r => data.callback?.(r));
const data = raw as {
adapter: string;
recipient: string;
callback?: (result: { ok: boolean; error?: string }) => void;
};
registry
.send({
adapter: data.adapter,
recipient: data.recipient ?? "",
text: `🏓 pi-channels test — ${new Date().toISOString()}`,
source: "channel:test",
})
.then((r) => data.callback?.(r));
});
}

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

@ -35,12 +35,12 @@
*/
import type { ExtensionAPI } from "@mariozechner/pi-coding-agent";
import { loadConfig } from "./config.ts";
import { ChannelRegistry } from "./registry.ts";
import { registerChannelEvents, setBridge } from "./events.ts";
import { registerChannelTool } from "./tool.ts";
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";
export default function (pi: ExtensionAPI) {
const log = createLogger(pi);
@ -76,7 +76,7 @@ export default function (pi: ExtensionAPI) {
// Start incoming/bidirectional adapters
await registry.startListening();
const startErrors = registry.getErrors().filter(e => e.error.startsWith("Failed to start"));
const startErrors = registry.getErrors().filter((e) => e.error.startsWith("Failed to start"));
for (const err of startErrors) {
ctx.ui.notify(`pi-channels: ${err.adapter}: ${err.error}`, "warning");
}
@ -105,9 +105,7 @@ export default function (pi: ExtensionAPI) {
pi.registerCommand("chat-bridge", {
description: "Manage chat bridge: /chat-bridge [on|off|status]",
getArgumentCompletions: (prefix: string) => {
return ["on", "off", "status"]
.filter(c => c.startsWith(prefix))
.map(c => ({ value: c, label: c }));
return ["on", "off", "status"].filter((c) => c.startsWith(prefix)).map((c) => ({ value: c, label: c }));
},
handler: async (args, ctx) => {
const cmd = args?.trim().toLowerCase();

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

@ -2,10 +2,18 @@
* pi-channels — Adapter registry + route resolution.
*/
import type { ChannelAdapter, ChannelMessage, AdapterConfig, ChannelConfig, AdapterDirection, OnIncomingMessage, IncomingMessage } from "./types.ts";
import { createSlackAdapter } from "./adapters/slack.ts";
import { createTelegramAdapter } from "./adapters/telegram.ts";
import { createWebhookAdapter } from "./adapters/webhook.ts";
import { createSlackAdapter } from "./adapters/slack.ts";
import type {
AdapterConfig,
AdapterDirection,
ChannelAdapter,
ChannelConfig,
ChannelMessage,
IncomingMessage,
OnIncomingMessage,
} from "./types.ts";
// ── Built-in adapter factories ──────────────────────────────────
@ -161,7 +169,8 @@ export class ChannelRegistry {
/** List all registered adapters and route aliases. */
list(): Array<{ name: string; type: "adapter" | "route"; direction?: AdapterDirection; target?: string }> {
const result: Array<{ name: string; type: "adapter" | "route"; direction?: AdapterDirection; target?: string }> = [];
const result: Array<{ name: string; type: "adapter" | "route"; direction?: AdapterDirection; target?: string }> =
[];
for (const [name, adapter] of this.adapters) {
result.push({ name, type: "adapter", direction: adapter.direction });
}

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

@ -2,9 +2,9 @@
* pi-channels — LLM tool registration.
*/
import { StringEnum } from "@mariozechner/pi-ai";
import type { ExtensionAPI } from "@mariozechner/pi-coding-agent";
import { Type } from "@sinclair/typebox";
import { StringEnum } from "@mariozechner/pi-ai";
import type { ChannelRegistry } from "./registry.ts";
interface ChannelToolParams {
@ -23,22 +23,15 @@ export function registerChannelTool(pi: ExtensionAPI, registry: ChannelRegistry)
"Send notifications via configured adapters (Telegram, webhooks, custom). " +
"Actions: send (deliver a message), list (show adapters + routes), test (send a ping).",
parameters: Type.Object({
action: StringEnum(
["send", "list", "test"] as const,
{ description: "Action to perform" },
) as any,
adapter: Type.Optional(
Type.String({ description: "Adapter name or route alias (required for send, test)" }),
),
action: StringEnum(["send", "list", "test"] as const, { description: "Action to perform" }) as any,
adapter: Type.Optional(Type.String({ description: "Adapter name or route alias (required for send, test)" })),
recipient: Type.Optional(
Type.String({ description: "Recipient — chat ID, webhook URL, etc. (required for send unless using a route)" }),
),
text: Type.Optional(
Type.String({ description: "Message text (required for send)" }),
),
source: Type.Optional(
Type.String({ description: "Source label (optional)" }),
Type.String({
description: "Recipient — chat ID, webhook URL, etc. (required for send unless using a route)",
}),
),
text: Type.Optional(Type.String({ description: "Message text (required for send)" })),
source: Type.Optional(Type.String({ description: "Source label (optional)" })),
}) as any,
async execute(_toolCallId, _params) {
@ -51,10 +44,10 @@ export function registerChannelTool(pi: ExtensionAPI, registry: ChannelRegistry)
if (items.length === 0) {
result = 'No adapters configured. Add "pi-channels" to your settings.json.';
} else {
const lines = items.map(i =>
const lines = items.map((i) =>
i.type === "route"
? `- **${i.name}** (route → ${i.target})`
: `- **${i.name}** (${i.direction ?? "adapter"})`
: `- **${i.name}** (${i.direction ?? "adapter"})`,
);
result = `**Channel (${items.length}):**\n${lines.join("\n")}`;
}