feat: listen mode (#997)
Co-authored-by: cpacker <packercharles@gmail.com>
This commit is contained in:
Shubham Naik
@@ -6223,6 +6223,54 @@ export default function App({
|
||||
return { submitted: true };
|
||||
}
|
||||
|
||||
// Special handling for /listen command - start listener mode
|
||||
if (trimmed === "/listen" || trimmed.startsWith("/listen ")) {
|
||||
// Tokenize with quote support: --name "my laptop"
|
||||
const parts = Array.from(
|
||||
trimmed.matchAll(
|
||||
/"([^"\\]*(?:\\.[^"\\]*)*)"|'([^'\\]*(?:\\.[^'\\]*)*)'|(\S+)/g,
|
||||
),
|
||||
(match) => match[1] ?? match[2] ?? match[3],
|
||||
);
|
||||
|
||||
let name: string | undefined;
|
||||
let listenAgentId: string | undefined;
|
||||
|
||||
for (let i = 1; i < parts.length; i++) {
|
||||
const part = parts[i];
|
||||
const nextPart = parts[i + 1];
|
||||
if (part === "--name" && nextPart) {
|
||||
name = nextPart;
|
||||
i++;
|
||||
} else if (part === "--agent" && nextPart) {
|
||||
listenAgentId = nextPart;
|
||||
i++;
|
||||
}
|
||||
}
|
||||
|
||||
// Default to current agent if not specified
|
||||
const targetAgentId = listenAgentId || agentId;
|
||||
|
||||
const cmd = commandRunner.start(msg, "Starting listener...");
|
||||
const { handleListen, setActiveCommandId: setActiveListenCommandId } =
|
||||
await import("./commands/listen");
|
||||
setActiveListenCommandId(cmd.id);
|
||||
try {
|
||||
await handleListen(
|
||||
{
|
||||
buffersRef,
|
||||
refreshDerived,
|
||||
setCommandRunning,
|
||||
},
|
||||
msg,
|
||||
{ name, agentId: targetAgentId },
|
||||
);
|
||||
} finally {
|
||||
setActiveListenCommandId(null);
|
||||
}
|
||||
return { submitted: true };
|
||||
}
|
||||
|
||||
// Special handling for /help command - opens help dialog
|
||||
if (trimmed === "/help") {
|
||||
startOverlayCommand(
|
||||
|
||||
335
src/cli/commands/listen.ts
Normal file
335
src/cli/commands/listen.ts
Normal file
@@ -0,0 +1,335 @@
|
||||
/**
|
||||
* Listen mode - Register letta-code as a listener to receive messages from Letta Cloud
|
||||
* Usage: letta listen --name "george"
|
||||
*/
|
||||
|
||||
import { hostname } from "node:os";
|
||||
import { getServerUrl } from "../../agent/client";
|
||||
import { settingsManager } from "../../settings-manager";
|
||||
import { getErrorMessage } from "../../utils/error";
|
||||
import type { Buffers, Line } from "../helpers/accumulator";
|
||||
|
||||
// tiny helper for unique ids
|
||||
function uid(prefix: string) {
|
||||
return `${prefix}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`;
|
||||
}
|
||||
|
||||
// Helper type for command result
|
||||
type CommandLine = Extract<Line, { kind: "command" }>;
|
||||
|
||||
let activeCommandId: string | null = null;
|
||||
|
||||
export function setActiveCommandId(id: string | null): void {
|
||||
activeCommandId = id;
|
||||
}
|
||||
|
||||
// Context passed to listen handler
|
||||
export interface ListenCommandContext {
|
||||
buffersRef: { current: Buffers };
|
||||
refreshDerived: () => void;
|
||||
setCommandRunning: (running: boolean) => void;
|
||||
}
|
||||
|
||||
// Helper to add a command result to buffers
|
||||
function addCommandResult(
|
||||
buffersRef: { current: Buffers },
|
||||
refreshDerived: () => void,
|
||||
input: string,
|
||||
output: string,
|
||||
success: boolean,
|
||||
phase: "running" | "finished" = "finished",
|
||||
): string {
|
||||
const cmdId = activeCommandId ?? uid("cmd");
|
||||
const existing = buffersRef.current.byId.get(cmdId);
|
||||
const nextInput =
|
||||
existing && existing.kind === "command" ? existing.input : input;
|
||||
const line: CommandLine = {
|
||||
kind: "command",
|
||||
id: cmdId,
|
||||
input: nextInput,
|
||||
output,
|
||||
phase,
|
||||
...(phase === "finished" && { success }),
|
||||
};
|
||||
buffersRef.current.byId.set(cmdId, line);
|
||||
if (!buffersRef.current.order.includes(cmdId)) {
|
||||
buffersRef.current.order.push(cmdId);
|
||||
}
|
||||
refreshDerived();
|
||||
return cmdId;
|
||||
}
|
||||
|
||||
// Helper to update an existing command result
|
||||
function updateCommandResult(
|
||||
buffersRef: { current: Buffers },
|
||||
refreshDerived: () => void,
|
||||
cmdId: string,
|
||||
input: string,
|
||||
output: string,
|
||||
success: boolean,
|
||||
phase: "running" | "finished" = "finished",
|
||||
): void {
|
||||
const existing = buffersRef.current.byId.get(cmdId);
|
||||
const nextInput =
|
||||
existing && existing.kind === "command" ? existing.input : input;
|
||||
const line: CommandLine = {
|
||||
kind: "command",
|
||||
id: cmdId,
|
||||
input: nextInput,
|
||||
output,
|
||||
phase,
|
||||
...(phase === "finished" && { success }),
|
||||
};
|
||||
buffersRef.current.byId.set(cmdId, line);
|
||||
refreshDerived();
|
||||
}
|
||||
|
||||
interface ListenOptions {
|
||||
name?: string;
|
||||
agentId?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle /listen command
|
||||
* Usage: /listen --name "george" [--agent agent-xyz]
|
||||
*/
|
||||
export async function handleListen(
|
||||
ctx: ListenCommandContext,
|
||||
msg: string,
|
||||
opts: ListenOptions = {},
|
||||
): Promise<void> {
|
||||
// Show usage if needed
|
||||
if (msg.includes("--help") || msg.includes("-h")) {
|
||||
addCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
msg,
|
||||
"Usage: /listen --name <connection-name> [--agent <agent-id>]\n\n" +
|
||||
"Register this letta-code instance to receive messages from Letta Cloud.\n\n" +
|
||||
"Options:\n" +
|
||||
" --name <name> Friendly name for this connection (required)\n" +
|
||||
" --agent <id> Bind connection to specific agent (defaults to current agent)\n\n" +
|
||||
"Examples:\n" +
|
||||
' /listen --name "george" # Uses current agent\n' +
|
||||
' /listen --name "laptop-work" --agent agent-abc123\n\n' +
|
||||
"Once connected, this instance will listen for incoming messages from cloud agents.\n" +
|
||||
"Messages will be executed locally using your letta-code environment.",
|
||||
true,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Validate required parameters
|
||||
const connectionName = opts.name;
|
||||
const agentId = opts.agentId;
|
||||
|
||||
if (!connectionName) {
|
||||
addCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
msg,
|
||||
"Error: --name is required\n\n" +
|
||||
'Usage: /listen --name "george"\n\n' +
|
||||
"Provide a friendly name to identify this connection (e.g., your name, device name).",
|
||||
false,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!agentId) {
|
||||
addCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
msg,
|
||||
"Error: No agent specified\n\n" +
|
||||
"This connection needs a default agent to execute messages.\n" +
|
||||
"If you're seeing this, it means no agent is active in this conversation.\n\n" +
|
||||
"Please start a conversation with an agent first, or specify one explicitly:\n" +
|
||||
' /listen --name "george" --agent agent-abc123',
|
||||
false,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Start listen flow
|
||||
ctx.setCommandRunning(true);
|
||||
|
||||
const cmdId = addCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
msg,
|
||||
"Connecting to Letta Cloud...",
|
||||
true,
|
||||
"running",
|
||||
);
|
||||
|
||||
try {
|
||||
// Get device ID (stable across sessions)
|
||||
const deviceId = settingsManager.getOrCreateDeviceId();
|
||||
const deviceName = hostname();
|
||||
|
||||
updateCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
cmdId,
|
||||
msg,
|
||||
`Registering listener "${connectionName}"...\n` +
|
||||
`Device: ${deviceName} (${deviceId.slice(0, 8)}...)`,
|
||||
true,
|
||||
"running",
|
||||
);
|
||||
|
||||
// Register with cloud to get connectionId
|
||||
const serverUrl = getServerUrl();
|
||||
const settings = await settingsManager.getSettingsWithSecureTokens();
|
||||
const apiKey = process.env.LETTA_API_KEY || settings.env?.LETTA_API_KEY;
|
||||
|
||||
if (!apiKey) {
|
||||
throw new Error("Missing LETTA_API_KEY");
|
||||
}
|
||||
|
||||
// Call register endpoint
|
||||
const registerUrl = `${serverUrl}/v1/listeners/register`;
|
||||
const registerResponse = await fetch(registerUrl, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
Authorization: `Bearer ${apiKey}`,
|
||||
"X-Letta-Source": "letta-code",
|
||||
},
|
||||
body: JSON.stringify({
|
||||
deviceId,
|
||||
connectionName,
|
||||
agentId: opts.agentId,
|
||||
}),
|
||||
});
|
||||
|
||||
if (!registerResponse.ok) {
|
||||
const error = (await registerResponse.json()) as { message?: string };
|
||||
throw new Error(error.message || "Registration failed");
|
||||
}
|
||||
|
||||
const { connectionId, wsUrl } = (await registerResponse.json()) as {
|
||||
connectionId: string;
|
||||
wsUrl: string;
|
||||
};
|
||||
|
||||
// Build agent info message
|
||||
const adeUrl = `https://app.letta.com/agents/${agentId}`;
|
||||
const agentInfo = `Agent: ${agentId}\n→ ${adeUrl}\n\n`;
|
||||
|
||||
updateCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
cmdId,
|
||||
msg,
|
||||
`✓ Registered successfully!\n\n` +
|
||||
`Connection ID: ${connectionId}\n` +
|
||||
`Name: "${connectionName}"\n` +
|
||||
agentInfo +
|
||||
`WebSocket: ${wsUrl}\n\n` +
|
||||
`Starting WebSocket connection...`,
|
||||
true,
|
||||
"running",
|
||||
);
|
||||
|
||||
// Import and start WebSocket client
|
||||
const { startListenerClient } = await import(
|
||||
"../../websocket/listen-client"
|
||||
);
|
||||
|
||||
await startListenerClient({
|
||||
connectionId,
|
||||
wsUrl,
|
||||
deviceId,
|
||||
connectionName,
|
||||
agentId,
|
||||
onStatusChange: (status, connId) => {
|
||||
const adeUrl = `https://app.letta.com/agents/${agentId}?deviceId=${connId}`;
|
||||
const statusText =
|
||||
status === "receiving"
|
||||
? "Receiving message"
|
||||
: status === "processing"
|
||||
? "Processing message"
|
||||
: "Awaiting instructions";
|
||||
|
||||
updateCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
cmdId,
|
||||
msg,
|
||||
`Connected to Letta Cloud\n` +
|
||||
`${statusText}\n\n` +
|
||||
`View in ADE → ${adeUrl}`,
|
||||
true,
|
||||
"finished",
|
||||
);
|
||||
},
|
||||
onRetrying: (attempt, _maxAttempts, nextRetryIn) => {
|
||||
const adeUrl = `https://app.letta.com/agents/${agentId}?deviceId=${connectionId}`;
|
||||
updateCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
cmdId,
|
||||
msg,
|
||||
`Reconnecting to Letta Cloud...\n` +
|
||||
`Attempt ${attempt}, retrying in ${Math.round(nextRetryIn / 1000)}s\n\n` +
|
||||
`View in ADE → ${adeUrl}`,
|
||||
true,
|
||||
"running",
|
||||
);
|
||||
},
|
||||
onConnected: () => {
|
||||
const adeUrl = `https://app.letta.com/agents/${agentId}?deviceId=${connectionId}`;
|
||||
|
||||
updateCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
cmdId,
|
||||
msg,
|
||||
`Connected to Letta Cloud\n` +
|
||||
`Awaiting instructions\n\n` +
|
||||
`View in ADE → ${adeUrl}`,
|
||||
true,
|
||||
"finished",
|
||||
);
|
||||
ctx.setCommandRunning(false);
|
||||
},
|
||||
onDisconnected: () => {
|
||||
updateCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
cmdId,
|
||||
msg,
|
||||
`✗ Listener disconnected\n\n` + `Connection to Letta Cloud was lost.`,
|
||||
false,
|
||||
"finished",
|
||||
);
|
||||
ctx.setCommandRunning(false);
|
||||
},
|
||||
onError: (error: Error) => {
|
||||
updateCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
cmdId,
|
||||
msg,
|
||||
`✗ Listener error: ${getErrorMessage(error)}`,
|
||||
false,
|
||||
"finished",
|
||||
);
|
||||
ctx.setCommandRunning(false);
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
updateCommandResult(
|
||||
ctx.buffersRef,
|
||||
ctx.refreshDerived,
|
||||
cmdId,
|
||||
msg,
|
||||
`✗ Failed to start listener: ${getErrorMessage(error)}`,
|
||||
false,
|
||||
"finished",
|
||||
);
|
||||
ctx.setCommandRunning(false);
|
||||
}
|
||||
}
|
||||
77
src/cli/components/ListenerStatusUI.tsx
Normal file
77
src/cli/components/ListenerStatusUI.tsx
Normal file
@@ -0,0 +1,77 @@
|
||||
import { Box, Text } from "ink";
|
||||
import Spinner from "ink-spinner";
|
||||
import { useEffect, useState } from "react";
|
||||
|
||||
interface ListenerStatusUIProps {
|
||||
agentId: string;
|
||||
connectionId: string;
|
||||
onReady: (callbacks: {
|
||||
updateStatus: (status: "idle" | "receiving" | "processing") => void;
|
||||
updateRetryStatus: (attempt: number, nextRetryIn: number) => void;
|
||||
clearRetryStatus: () => void;
|
||||
}) => void;
|
||||
}
|
||||
|
||||
export function ListenerStatusUI(props: ListenerStatusUIProps) {
|
||||
const { agentId, connectionId, onReady } = props;
|
||||
const [status, setStatus] = useState<"idle" | "receiving" | "processing">(
|
||||
"idle",
|
||||
);
|
||||
const [retryInfo, setRetryInfo] = useState<{
|
||||
attempt: number;
|
||||
nextRetryIn: number;
|
||||
} | null>(null);
|
||||
|
||||
useEffect(() => {
|
||||
onReady({
|
||||
updateStatus: setStatus,
|
||||
updateRetryStatus: (attempt, nextRetryIn) => {
|
||||
setRetryInfo({ attempt, nextRetryIn });
|
||||
},
|
||||
clearRetryStatus: () => {
|
||||
setRetryInfo(null);
|
||||
},
|
||||
});
|
||||
}, [onReady]);
|
||||
|
||||
const adeUrl = `https://app.letta.com/agents/${agentId}?deviceId=${connectionId}`;
|
||||
|
||||
const statusText = retryInfo
|
||||
? `Reconnecting (attempt ${retryInfo.attempt}, retry in ${Math.round(retryInfo.nextRetryIn / 1000)}s)`
|
||||
: status === "receiving"
|
||||
? "Receiving message"
|
||||
: status === "processing"
|
||||
? "Processing message"
|
||||
: "Awaiting instructions";
|
||||
|
||||
const showSpinner = status !== "idle" || retryInfo !== null;
|
||||
|
||||
return (
|
||||
<Box flexDirection="column" paddingX={1} paddingY={1}>
|
||||
<Box marginBottom={1}>
|
||||
<Text bold color="green">
|
||||
Connected to Letta Cloud
|
||||
</Text>
|
||||
</Box>
|
||||
|
||||
<Box marginBottom={1}>
|
||||
{showSpinner && (
|
||||
<Text>
|
||||
<Text color={retryInfo ? "yellow" : "cyan"}>
|
||||
<Spinner type="dots" />
|
||||
</Text>{" "}
|
||||
<Text color={retryInfo ? "yellow" : undefined}>{statusText}</Text>
|
||||
</Text>
|
||||
)}
|
||||
{!showSpinner && <Text dimColor>{statusText}</Text>}
|
||||
</Box>
|
||||
|
||||
<Box>
|
||||
<Text dimColor>View in ADE → </Text>
|
||||
<Text color="blue" underline>
|
||||
{adeUrl}
|
||||
</Text>
|
||||
</Box>
|
||||
</Box>
|
||||
);
|
||||
}
|
||||
187
src/cli/subcommands/listen.tsx
Normal file
187
src/cli/subcommands/listen.tsx
Normal file
@@ -0,0 +1,187 @@
|
||||
/**
|
||||
* CLI subcommand: letta listen --name "george"
|
||||
* Register letta-code as a listener to receive messages from Letta Cloud
|
||||
*/
|
||||
|
||||
import { parseArgs } from "node:util";
|
||||
import { render } from "ink";
|
||||
import { getServerUrl } from "../../agent/client";
|
||||
import { settingsManager } from "../../settings-manager";
|
||||
import { ListenerStatusUI } from "../components/ListenerStatusUI";
|
||||
|
||||
export async function runListenSubcommand(argv: string[]): Promise<number> {
|
||||
// Parse arguments
|
||||
const { values } = parseArgs({
|
||||
args: argv,
|
||||
options: {
|
||||
name: { type: "string" },
|
||||
agent: { type: "string" },
|
||||
help: { type: "boolean", short: "h" },
|
||||
},
|
||||
allowPositionals: false,
|
||||
});
|
||||
|
||||
// Show help
|
||||
if (values.help) {
|
||||
console.log(
|
||||
"Usage: letta listen --name <connection-name> [--agent <agent-id>]\n",
|
||||
);
|
||||
console.log(
|
||||
"Register this letta-code instance to receive messages from Letta Cloud.\n",
|
||||
);
|
||||
console.log("Options:");
|
||||
console.log(
|
||||
" --name <name> Friendly name for this connection (required)",
|
||||
);
|
||||
console.log(
|
||||
" --agent <id> Bind connection to specific agent (required for CLI usage)",
|
||||
);
|
||||
console.log(" -h, --help Show this help message\n");
|
||||
console.log("Examples:");
|
||||
console.log(' letta listen --name "george" --agent agent-abc123');
|
||||
console.log(' letta listen --name "laptop-work" --agent agent-xyz789\n');
|
||||
console.log(
|
||||
"Once connected, this instance will listen for incoming messages from cloud agents.",
|
||||
);
|
||||
console.log(
|
||||
"Messages will be executed locally using your letta-code environment.",
|
||||
);
|
||||
return 0;
|
||||
}
|
||||
|
||||
const connectionName = values.name;
|
||||
const agentId = values.agent;
|
||||
|
||||
if (!connectionName) {
|
||||
console.error("Error: --name is required\n");
|
||||
console.error('Usage: letta listen --name "george" --agent agent-abc123\n');
|
||||
console.error(
|
||||
"Provide a friendly name to identify this connection (e.g., your name, device name).",
|
||||
);
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (!agentId) {
|
||||
console.error("Error: --agent is required\n");
|
||||
console.error('Usage: letta listen --name "george" --agent agent-abc123\n');
|
||||
console.error(
|
||||
"A listener connection needs a default agent to execute messages.",
|
||||
);
|
||||
console.error(
|
||||
"Specify which agent should receive messages from this connection.",
|
||||
);
|
||||
return 1;
|
||||
}
|
||||
|
||||
try {
|
||||
// Get device ID
|
||||
const deviceId = settingsManager.getOrCreateDeviceId();
|
||||
|
||||
// Get API key (include secure token storage fallback)
|
||||
const settings = await settingsManager.getSettingsWithSecureTokens();
|
||||
const apiKey = process.env.LETTA_API_KEY || settings.env?.LETTA_API_KEY;
|
||||
|
||||
if (!apiKey) {
|
||||
console.error("Error: LETTA_API_KEY not found");
|
||||
console.error("Set your API key with: export LETTA_API_KEY=<your-key>");
|
||||
return 1;
|
||||
}
|
||||
|
||||
// Register with cloud
|
||||
const serverUrl = getServerUrl();
|
||||
const registerUrl = `${serverUrl}/v1/listeners/register`;
|
||||
|
||||
const registerResponse = await fetch(registerUrl, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
Authorization: `Bearer ${apiKey}`,
|
||||
"X-Letta-Source": "letta-code",
|
||||
},
|
||||
body: JSON.stringify({
|
||||
deviceId,
|
||||
connectionName,
|
||||
agentId,
|
||||
}),
|
||||
});
|
||||
|
||||
if (!registerResponse.ok) {
|
||||
const error = (await registerResponse.json()) as { message?: string };
|
||||
console.error(`Registration failed: ${error.message || "Unknown error"}`);
|
||||
return 1;
|
||||
}
|
||||
|
||||
const { connectionId, wsUrl } = (await registerResponse.json()) as {
|
||||
connectionId: string;
|
||||
wsUrl: string;
|
||||
};
|
||||
|
||||
// Clear screen and render Ink UI
|
||||
console.clear();
|
||||
|
||||
let updateStatusCallback:
|
||||
| ((status: "idle" | "receiving" | "processing") => void)
|
||||
| null = null;
|
||||
let updateRetryStatusCallback:
|
||||
| ((attempt: number, nextRetryIn: number) => void)
|
||||
| null = null;
|
||||
let clearRetryStatusCallback: (() => void) | null = null;
|
||||
|
||||
const { unmount } = render(
|
||||
<ListenerStatusUI
|
||||
agentId={agentId}
|
||||
connectionId={connectionId}
|
||||
onReady={(callbacks) => {
|
||||
updateStatusCallback = callbacks.updateStatus;
|
||||
updateRetryStatusCallback = callbacks.updateRetryStatus;
|
||||
clearRetryStatusCallback = callbacks.clearRetryStatus;
|
||||
}}
|
||||
/>,
|
||||
);
|
||||
|
||||
// Import and start WebSocket client
|
||||
const { startListenerClient } = await import(
|
||||
"../../websocket/listen-client"
|
||||
);
|
||||
|
||||
await startListenerClient({
|
||||
connectionId,
|
||||
wsUrl,
|
||||
deviceId,
|
||||
connectionName,
|
||||
agentId,
|
||||
onStatusChange: (status) => {
|
||||
clearRetryStatusCallback?.();
|
||||
updateStatusCallback?.(status);
|
||||
},
|
||||
onConnected: () => {
|
||||
clearRetryStatusCallback?.();
|
||||
updateStatusCallback?.("idle");
|
||||
},
|
||||
onRetrying: (attempt, _maxAttempts, nextRetryIn) => {
|
||||
updateRetryStatusCallback?.(attempt, nextRetryIn);
|
||||
},
|
||||
onDisconnected: () => {
|
||||
unmount();
|
||||
console.log("\n✗ Listener disconnected");
|
||||
console.log("Connection to Letta Cloud was lost.\n");
|
||||
process.exit(1);
|
||||
},
|
||||
onError: (error: Error) => {
|
||||
unmount();
|
||||
console.error(`\n✗ Listener error: ${error.message}\n`);
|
||||
process.exit(1);
|
||||
},
|
||||
});
|
||||
|
||||
// Keep process alive
|
||||
return new Promise<number>(() => {
|
||||
// Never resolves - runs until Ctrl+C
|
||||
});
|
||||
} catch (error) {
|
||||
console.error(
|
||||
`Failed to start listener: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { runAgentsSubcommand } from "./agents";
|
||||
import { runBlocksSubcommand } from "./blocks";
|
||||
import { runListenSubcommand } from "./listen.tsx";
|
||||
import { runMemfsSubcommand } from "./memfs";
|
||||
import { runMessagesSubcommand } from "./messages";
|
||||
|
||||
@@ -19,6 +20,8 @@ export async function runSubcommand(argv: string[]): Promise<number | null> {
|
||||
return runMessagesSubcommand(rest);
|
||||
case "blocks":
|
||||
return runBlocksSubcommand(rest);
|
||||
case "listen":
|
||||
return runListenSubcommand(rest);
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
|
||||
552
src/websocket/listen-client.ts
Normal file
552
src/websocket/listen-client.ts
Normal file
@@ -0,0 +1,552 @@
|
||||
/**
|
||||
* WebSocket client for listen mode
|
||||
* Connects to Letta Cloud and receives messages to execute locally
|
||||
*/
|
||||
|
||||
import type { Stream } from "@letta-ai/letta-client/core/streaming";
|
||||
import type { MessageCreate } from "@letta-ai/letta-client/resources/agents/agents";
|
||||
import type {
|
||||
ApprovalCreate,
|
||||
LettaStreamingResponse,
|
||||
ToolReturn,
|
||||
} from "@letta-ai/letta-client/resources/agents/messages";
|
||||
import WebSocket from "ws";
|
||||
import {
|
||||
type ApprovalDecision,
|
||||
type ApprovalResult,
|
||||
executeApprovalBatch,
|
||||
} from "../agent/approval-execution";
|
||||
import { getResumeData } from "../agent/check-approval";
|
||||
import { getClient } from "../agent/client";
|
||||
import { sendMessageStream } from "../agent/message";
|
||||
import { createBuffers } from "../cli/helpers/accumulator";
|
||||
import { drainStreamWithResume } from "../cli/helpers/stream";
|
||||
import { settingsManager } from "../settings-manager";
|
||||
import { loadTools } from "../tools/manager";
|
||||
|
||||
interface StartListenerOptions {
|
||||
connectionId: string;
|
||||
wsUrl: string;
|
||||
deviceId: string;
|
||||
connectionName: string;
|
||||
agentId?: string;
|
||||
onConnected: () => void;
|
||||
onDisconnected: () => void;
|
||||
onError: (error: Error) => void;
|
||||
onStatusChange?: (
|
||||
status: "idle" | "receiving" | "processing",
|
||||
connectionId: string,
|
||||
) => void;
|
||||
onRetrying?: (
|
||||
attempt: number,
|
||||
maxAttempts: number,
|
||||
nextRetryIn: number,
|
||||
) => void;
|
||||
}
|
||||
|
||||
interface PingMessage {
|
||||
type: "ping";
|
||||
}
|
||||
|
||||
interface PongMessage {
|
||||
type: "pong";
|
||||
}
|
||||
|
||||
interface IncomingMessage {
|
||||
type: "message";
|
||||
agentId?: string;
|
||||
conversationId?: string;
|
||||
messages: Array<MessageCreate | ApprovalCreate>;
|
||||
}
|
||||
|
||||
interface ResultMessage {
|
||||
type: "result";
|
||||
success: boolean;
|
||||
stopReason?: string;
|
||||
}
|
||||
|
||||
interface RunStartedMessage {
|
||||
type: "run_started";
|
||||
runId: string;
|
||||
}
|
||||
|
||||
type ServerMessage = PongMessage | IncomingMessage;
|
||||
type ClientMessage = PingMessage | ResultMessage | RunStartedMessage;
|
||||
|
||||
type ListenerRuntime = {
|
||||
socket: WebSocket | null;
|
||||
heartbeatInterval: NodeJS.Timeout | null;
|
||||
reconnectTimeout: NodeJS.Timeout | null;
|
||||
intentionallyClosed: boolean;
|
||||
hasSuccessfulConnection: boolean;
|
||||
messageQueue: Promise<void>;
|
||||
};
|
||||
|
||||
type ApprovalSlot =
|
||||
| { type: "result"; value: ApprovalResult }
|
||||
| { type: "decision" };
|
||||
|
||||
// Listen mode supports one active connection per process.
|
||||
let activeRuntime: ListenerRuntime | null = null;
|
||||
|
||||
const MAX_RETRY_DURATION_MS = 5 * 60 * 1000; // 5 minutes
|
||||
const INITIAL_RETRY_DELAY_MS = 1000; // 1 second
|
||||
const MAX_RETRY_DELAY_MS = 30000; // 30 seconds
|
||||
|
||||
function createRuntime(): ListenerRuntime {
|
||||
return {
|
||||
socket: null,
|
||||
heartbeatInterval: null,
|
||||
reconnectTimeout: null,
|
||||
intentionallyClosed: false,
|
||||
hasSuccessfulConnection: false,
|
||||
messageQueue: Promise.resolve(),
|
||||
};
|
||||
}
|
||||
|
||||
function clearRuntimeTimers(runtime: ListenerRuntime): void {
|
||||
if (runtime.reconnectTimeout) {
|
||||
clearTimeout(runtime.reconnectTimeout);
|
||||
runtime.reconnectTimeout = null;
|
||||
}
|
||||
if (runtime.heartbeatInterval) {
|
||||
clearInterval(runtime.heartbeatInterval);
|
||||
runtime.heartbeatInterval = null;
|
||||
}
|
||||
}
|
||||
|
||||
function stopRuntime(
|
||||
runtime: ListenerRuntime,
|
||||
suppressCallbacks: boolean,
|
||||
): void {
|
||||
runtime.intentionallyClosed = true;
|
||||
clearRuntimeTimers(runtime);
|
||||
|
||||
if (!runtime.socket) {
|
||||
return;
|
||||
}
|
||||
|
||||
const socket = runtime.socket;
|
||||
runtime.socket = null;
|
||||
|
||||
// Stale runtimes being replaced should not emit callbacks/retries.
|
||||
if (suppressCallbacks) {
|
||||
socket.removeAllListeners();
|
||||
}
|
||||
|
||||
if (
|
||||
socket.readyState === WebSocket.OPEN ||
|
||||
socket.readyState === WebSocket.CONNECTING
|
||||
) {
|
||||
socket.close();
|
||||
}
|
||||
}
|
||||
|
||||
function parseServerMessage(data: WebSocket.RawData): ServerMessage | null {
|
||||
try {
|
||||
const raw = typeof data === "string" ? data : data.toString();
|
||||
const parsed = JSON.parse(raw) as { type?: string };
|
||||
if (parsed.type === "pong" || parsed.type === "message") {
|
||||
return parsed as ServerMessage;
|
||||
}
|
||||
return null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function sendClientMessage(socket: WebSocket, payload: ClientMessage): void {
|
||||
if (socket.readyState === WebSocket.OPEN) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
}
|
||||
}
|
||||
|
||||
function buildApprovalExecutionPlan(
|
||||
approvalMessage: ApprovalCreate,
|
||||
pendingApprovals: Array<{
|
||||
toolCallId: string;
|
||||
toolName: string;
|
||||
toolArgs: string;
|
||||
}>,
|
||||
): {
|
||||
slots: ApprovalSlot[];
|
||||
decisions: ApprovalDecision[];
|
||||
} {
|
||||
const pendingByToolCallId = new Map(
|
||||
pendingApprovals.map((approval) => [approval.toolCallId, approval]),
|
||||
);
|
||||
|
||||
const slots: ApprovalSlot[] = [];
|
||||
const decisions: ApprovalDecision[] = [];
|
||||
|
||||
for (const approval of approvalMessage.approvals ?? []) {
|
||||
if (approval.type === "tool") {
|
||||
slots.push({ type: "result", value: approval as ToolReturn });
|
||||
continue;
|
||||
}
|
||||
|
||||
if (approval.type !== "approval") {
|
||||
slots.push({
|
||||
type: "result",
|
||||
value: {
|
||||
type: "tool",
|
||||
tool_call_id: "unknown",
|
||||
tool_return: "Error: Unsupported approval payload",
|
||||
status: "error",
|
||||
},
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
const pending = pendingByToolCallId.get(approval.tool_call_id);
|
||||
|
||||
if (approval.approve) {
|
||||
if (!pending) {
|
||||
slots.push({
|
||||
type: "result",
|
||||
value: {
|
||||
type: "tool",
|
||||
tool_call_id: approval.tool_call_id,
|
||||
tool_return: "Error: Pending approval not found",
|
||||
status: "error",
|
||||
},
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
decisions.push({
|
||||
type: "approve",
|
||||
approval: {
|
||||
toolCallId: pending.toolCallId,
|
||||
toolName: pending.toolName,
|
||||
toolArgs: pending.toolArgs || "{}",
|
||||
},
|
||||
});
|
||||
slots.push({ type: "decision" });
|
||||
continue;
|
||||
}
|
||||
|
||||
decisions.push({
|
||||
type: "deny",
|
||||
approval: {
|
||||
toolCallId: approval.tool_call_id,
|
||||
toolName: pending?.toolName ?? "",
|
||||
toolArgs: pending?.toolArgs ?? "{}",
|
||||
},
|
||||
reason:
|
||||
typeof approval.reason === "string" && approval.reason.length > 0
|
||||
? approval.reason
|
||||
: "Tool execution denied",
|
||||
});
|
||||
slots.push({ type: "decision" });
|
||||
}
|
||||
|
||||
return { slots, decisions };
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the listener WebSocket client with automatic retry.
|
||||
*/
|
||||
export async function startListenerClient(
|
||||
opts: StartListenerOptions,
|
||||
): Promise<void> {
|
||||
// Replace any existing runtime without stale callback leakage.
|
||||
if (activeRuntime) {
|
||||
stopRuntime(activeRuntime, true);
|
||||
}
|
||||
|
||||
const runtime = createRuntime();
|
||||
activeRuntime = runtime;
|
||||
|
||||
await connectWithRetry(runtime, opts);
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect to WebSocket with exponential backoff retry.
|
||||
*/
|
||||
async function connectWithRetry(
|
||||
runtime: ListenerRuntime,
|
||||
opts: StartListenerOptions,
|
||||
attempt: number = 0,
|
||||
startTime: number = Date.now(),
|
||||
): Promise<void> {
|
||||
if (runtime !== activeRuntime || runtime.intentionallyClosed) {
|
||||
return;
|
||||
}
|
||||
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
if (attempt > 0) {
|
||||
if (elapsedTime >= MAX_RETRY_DURATION_MS) {
|
||||
opts.onError(new Error("Failed to connect after 5 minutes of retrying"));
|
||||
return;
|
||||
}
|
||||
|
||||
const delay = Math.min(
|
||||
INITIAL_RETRY_DELAY_MS * 2 ** (attempt - 1),
|
||||
MAX_RETRY_DELAY_MS,
|
||||
);
|
||||
const maxAttempts = Math.ceil(
|
||||
Math.log2(MAX_RETRY_DURATION_MS / INITIAL_RETRY_DELAY_MS),
|
||||
);
|
||||
|
||||
opts.onRetrying?.(attempt, maxAttempts, delay);
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
runtime.reconnectTimeout = setTimeout(resolve, delay);
|
||||
});
|
||||
|
||||
runtime.reconnectTimeout = null;
|
||||
if (runtime !== activeRuntime || runtime.intentionallyClosed) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
clearRuntimeTimers(runtime);
|
||||
|
||||
if (attempt === 0) {
|
||||
await loadTools();
|
||||
}
|
||||
|
||||
const settings = await settingsManager.getSettingsWithSecureTokens();
|
||||
const apiKey = process.env.LETTA_API_KEY || settings.env?.LETTA_API_KEY;
|
||||
|
||||
if (!apiKey) {
|
||||
throw new Error("Missing LETTA_API_KEY");
|
||||
}
|
||||
|
||||
const url = new URL(opts.wsUrl);
|
||||
url.searchParams.set("deviceId", opts.deviceId);
|
||||
url.searchParams.set("connectionName", opts.connectionName);
|
||||
if (opts.agentId) {
|
||||
url.searchParams.set("agentId", opts.agentId);
|
||||
}
|
||||
|
||||
const socket = new WebSocket(url.toString(), {
|
||||
headers: {
|
||||
Authorization: `Bearer ${apiKey}`,
|
||||
},
|
||||
});
|
||||
|
||||
runtime.socket = socket;
|
||||
|
||||
socket.on("open", () => {
|
||||
if (runtime !== activeRuntime || runtime.intentionallyClosed) {
|
||||
return;
|
||||
}
|
||||
|
||||
runtime.hasSuccessfulConnection = true;
|
||||
opts.onConnected();
|
||||
|
||||
runtime.heartbeatInterval = setInterval(() => {
|
||||
sendClientMessage(socket, { type: "ping" });
|
||||
}, 30000);
|
||||
});
|
||||
|
||||
socket.on("message", (data: WebSocket.RawData) => {
|
||||
const parsed = parseServerMessage(data);
|
||||
if (!parsed || parsed.type !== "message") {
|
||||
return;
|
||||
}
|
||||
|
||||
runtime.messageQueue = runtime.messageQueue
|
||||
.then(async () => {
|
||||
if (runtime !== activeRuntime || runtime.intentionallyClosed) {
|
||||
return;
|
||||
}
|
||||
|
||||
opts.onStatusChange?.("receiving", opts.connectionId);
|
||||
await handleIncomingMessage(
|
||||
parsed,
|
||||
socket,
|
||||
opts.onStatusChange,
|
||||
opts.connectionId,
|
||||
);
|
||||
opts.onStatusChange?.("idle", opts.connectionId);
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
if (process.env.DEBUG) {
|
||||
console.error("[Listen] Error handling queued message:", error);
|
||||
}
|
||||
opts.onStatusChange?.("idle", opts.connectionId);
|
||||
});
|
||||
});
|
||||
|
||||
socket.on("close", (code: number, reason: Buffer) => {
|
||||
if (runtime !== activeRuntime) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (process.env.DEBUG) {
|
||||
console.log(
|
||||
`[Listen] WebSocket disconnected (code: ${code}, reason: ${reason.toString()})`,
|
||||
);
|
||||
}
|
||||
|
||||
clearRuntimeTimers(runtime);
|
||||
runtime.socket = null;
|
||||
|
||||
if (runtime.intentionallyClosed) {
|
||||
opts.onDisconnected();
|
||||
return;
|
||||
}
|
||||
|
||||
// If we had connected before, restart backoff from zero for this outage window.
|
||||
const nextAttempt = runtime.hasSuccessfulConnection ? 0 : attempt + 1;
|
||||
const nextStartTime = runtime.hasSuccessfulConnection
|
||||
? Date.now()
|
||||
: startTime;
|
||||
runtime.hasSuccessfulConnection = false;
|
||||
|
||||
connectWithRetry(runtime, opts, nextAttempt, nextStartTime).catch(
|
||||
(error) => {
|
||||
opts.onError(error instanceof Error ? error : new Error(String(error)));
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
socket.on("error", (error: Error) => {
|
||||
if (process.env.DEBUG) {
|
||||
console.error("[Listen] WebSocket error:", error);
|
||||
}
|
||||
// Error triggers close(), which handles retry logic.
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle an incoming message from the cloud.
|
||||
*/
|
||||
async function handleIncomingMessage(
|
||||
msg: IncomingMessage,
|
||||
socket: WebSocket,
|
||||
onStatusChange?: (
|
||||
status: "idle" | "receiving" | "processing",
|
||||
connectionId: string,
|
||||
) => void,
|
||||
connectionId?: string,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const agentId = msg.agentId;
|
||||
const requestedConversationId = msg.conversationId;
|
||||
const conversationId = requestedConversationId ?? "default";
|
||||
|
||||
if (!agentId) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (connectionId) {
|
||||
onStatusChange?.("processing", connectionId);
|
||||
}
|
||||
|
||||
let messagesToSend: Array<MessageCreate | ApprovalCreate> = msg.messages;
|
||||
|
||||
const firstMessage = msg.messages[0];
|
||||
const isApprovalMessage =
|
||||
firstMessage &&
|
||||
"type" in firstMessage &&
|
||||
firstMessage.type === "approval" &&
|
||||
"approvals" in firstMessage;
|
||||
|
||||
if (isApprovalMessage) {
|
||||
const approvalMessage = firstMessage as ApprovalCreate;
|
||||
const client = await getClient();
|
||||
const agent = await client.agents.retrieve(agentId);
|
||||
const resumeData = await getResumeData(
|
||||
client,
|
||||
agent,
|
||||
requestedConversationId,
|
||||
);
|
||||
|
||||
const { slots, decisions } = buildApprovalExecutionPlan(
|
||||
approvalMessage,
|
||||
resumeData.pendingApprovals,
|
||||
);
|
||||
const decisionResults =
|
||||
decisions.length > 0 ? await executeApprovalBatch(decisions) : [];
|
||||
|
||||
const rebuiltApprovals: ApprovalResult[] = [];
|
||||
let decisionResultIndex = 0;
|
||||
|
||||
for (const slot of slots) {
|
||||
if (slot.type === "result") {
|
||||
rebuiltApprovals.push(slot.value);
|
||||
continue;
|
||||
}
|
||||
|
||||
const next = decisionResults[decisionResultIndex];
|
||||
if (next) {
|
||||
rebuiltApprovals.push(next);
|
||||
decisionResultIndex++;
|
||||
continue;
|
||||
}
|
||||
|
||||
rebuiltApprovals.push({
|
||||
type: "tool",
|
||||
tool_call_id: "unknown",
|
||||
tool_return: "Error: Missing approval execution result",
|
||||
status: "error",
|
||||
});
|
||||
}
|
||||
|
||||
messagesToSend = [
|
||||
{
|
||||
type: "approval",
|
||||
approvals: rebuiltApprovals,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
const stream = await sendMessageStream(conversationId, messagesToSend, {
|
||||
agentId,
|
||||
streamTokens: true,
|
||||
background: true,
|
||||
});
|
||||
|
||||
let runIdSent = false;
|
||||
|
||||
const buffers = createBuffers(agentId);
|
||||
const result = await drainStreamWithResume(
|
||||
stream as Stream<LettaStreamingResponse>,
|
||||
buffers,
|
||||
() => {},
|
||||
undefined,
|
||||
undefined,
|
||||
({ chunk }) => {
|
||||
const maybeRunId = (chunk as { run_id?: unknown }).run_id;
|
||||
if (!runIdSent && typeof maybeRunId === "string") {
|
||||
runIdSent = true;
|
||||
sendClientMessage(socket, {
|
||||
type: "run_started",
|
||||
runId: maybeRunId,
|
||||
});
|
||||
}
|
||||
return undefined;
|
||||
},
|
||||
);
|
||||
|
||||
sendClientMessage(socket, {
|
||||
type: "result",
|
||||
success: result.stopReason === "end_turn",
|
||||
stopReason: result.stopReason,
|
||||
});
|
||||
} catch {
|
||||
sendClientMessage(socket, {
|
||||
type: "result",
|
||||
success: false,
|
||||
stopReason: "error",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the active listener connection.
|
||||
*/
|
||||
export function stopListenerClient(): void {
|
||||
if (!activeRuntime) {
|
||||
return;
|
||||
}
|
||||
|
||||
const runtime = activeRuntime;
|
||||
activeRuntime = null;
|
||||
stopRuntime(runtime, true);
|
||||
}
|
||||
Reference in New Issue
Block a user