refactor: share IPC message types between CLI and daemon #94

Merged
xiaomo merged 2 commits from refactor/93-shared-ipc-types into main 2026-04-24 15:14:51 +00:00
9 changed files with 209 additions and 97 deletions
+3 -3
View File
@@ -514,7 +514,7 @@ describe("triggerWorkflowViaDaemon", () => {
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await triggerWorkflowViaDaemon(sockPath, "my-workflow", {});
const result = await triggerWorkflowViaDaemon(sockPath, "my-workflow", "", 100);
expect(result).toEqual({ ok: true });
} finally {
await new Promise<void>((r) => server.close(() => r()));
@@ -530,7 +530,7 @@ describe("triggerWorkflowViaDaemon", () => {
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await triggerWorkflowViaDaemon(sockPath, "missing", {});
const result = await triggerWorkflowViaDaemon(sockPath, "missing", "", 100);
expect(result).toEqual({ ok: false, error: "unknown workflow" });
} finally {
await new Promise<void>((r) => server.close(() => r()));
@@ -538,7 +538,7 @@ describe("triggerWorkflowViaDaemon", () => {
});
it("rejects when no daemon is listening on the socket", async () => {
await expect(triggerWorkflowViaDaemon(sockPath, "my-workflow", {})).rejects.toThrow(
await expect(triggerWorkflowViaDaemon(sockPath, "my-workflow", "", 100)).rejects.toThrow(
/Cannot connect to daemon/,
);
});
+14 -4
View File
@@ -1,7 +1,8 @@
import { existsSync } from "node:fs";
import { join } from "node:path";
import { isPlainRecord } from "@uncaged/nerve-core";
import type { DaemonIpcTriggerResponse } from "@uncaged/nerve-core";
import { DEFAULT_ENGINE_MAX_ROUNDS, isPlainRecord } from "@uncaged/nerve-core";
import { defineCommand } from "citty";
import { stringify } from "yaml";
@@ -513,7 +514,8 @@ const workflowTriggerCommand = defineCommand({
},
payload: {
type: "string",
description: "JSON payload to pass as trigger payload (default: {})",
description:
'JSON with optional "prompt" (string) and "maxRounds" (number) for the workflow run (default: {})',
default: "{}",
},
},
@@ -526,15 +528,23 @@ const workflowTriggerCommand = defineCommand({
process.exit(1);
}
let prompt = "";
let maxRounds = DEFAULT_ENGINE_MAX_ROUNDS;
if (isPlainRecord(triggerPayload)) {
const p = triggerPayload;
if (typeof p.prompt === "string") prompt = p.prompt;
if (typeof p.maxRounds === "number") maxRounds = p.maxRounds;
}
if (!isRunning()) {
process.stderr.write("❌ Nerve daemon is not running. Start it with `nerve start`.\n");
process.exit(1);
}
const socketPath = getSocketPath();
let response: { ok: true } | { ok: false; error: string };
let response: DaemonIpcTriggerResponse;
try {
response = await triggerWorkflowViaDaemon(socketPath, args.name, triggerPayload);
response = await triggerWorkflowViaDaemon(socketPath, args.name, prompt, maxRounds);
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`❌ Failed to contact daemon: ${msg}\n`);
+28 -19
View File
@@ -8,7 +8,12 @@
import { connect } from "node:net";
import type { Socket } from "node:net";
import type { SenseInfo } from "@uncaged/nerve-core";
import type {
DaemonIpcListSensesResponse,
DaemonIpcRequest,
DaemonIpcTriggerResponse,
SenseInfo,
} from "@uncaged/nerve-core";
import { isPlainRecord } from "@uncaged/nerve-core";
const CONNECT_TIMEOUT_MS = 3_000;
@@ -16,10 +21,6 @@ const RESPONSE_TIMEOUT_MS = 5_000;
export type { SenseInfo };
type TriggerResponse = { ok: true } | { ok: false; error: string };
type ListSensesResponse = { ok: true; senses: SenseInfo[] } | { ok: false; error: string };
function isSenseInfo(value: unknown): value is SenseInfo {
if (!isPlainRecord(value)) return false;
return (
@@ -31,7 +32,7 @@ function isSenseInfo(value: unknown): value is SenseInfo {
);
}
function parseDaemonResponse(line: string): TriggerResponse {
function parseDaemonResponse(line: string): DaemonIpcTriggerResponse {
try {
const obj: unknown = JSON.parse(line);
if (isPlainRecord(obj)) {
@@ -45,7 +46,7 @@ function parseDaemonResponse(line: string): TriggerResponse {
return { ok: false, error: `Unexpected daemon response: ${line}` };
}
function parseListSensesResponse(line: string): ListSensesResponse {
function parseListSensesResponse(line: string): DaemonIpcListSensesResponse {
try {
const obj: unknown = JSON.parse(line);
if (isPlainRecord(obj)) {
@@ -67,7 +68,7 @@ function parseListSensesResponse(line: string): ListSensesResponse {
*/
function sendAndReceive<T>(
socketPath: string,
message: object,
message: DaemonIpcRequest,
parseFirstLine: (trimmed: string) => T,
responseTimeoutMs: number = RESPONSE_TIMEOUT_MS,
): Promise<T> {
@@ -132,27 +133,35 @@ function sendAndReceive<T>(
export function triggerWorkflowViaDaemon(
socketPath: string,
workflow: string,
payload: unknown,
): Promise<TriggerResponse> {
return sendAndReceive(
socketPath,
{ type: "trigger-workflow", workflow, payload },
parseDaemonResponse,
);
prompt: string,
maxRounds: number,
): Promise<DaemonIpcTriggerResponse> {
const message: DaemonIpcRequest = {
type: "trigger-workflow",
workflow,
prompt,
maxRounds,
};
return sendAndReceive(socketPath, message, parseDaemonResponse);
}
/**
* Send a trigger-sense message to the running daemon via its Unix socket.
* Resolves with the daemon's response or rejects on connection/timeout errors.
*/
export function triggerSenseViaDaemon(socketPath: string, sense: string): Promise<TriggerResponse> {
return sendAndReceive(socketPath, { type: "trigger-sense", sense }, parseDaemonResponse);
export function triggerSenseViaDaemon(
socketPath: string,
sense: string,
): Promise<DaemonIpcTriggerResponse> {
const message: DaemonIpcRequest = { type: "trigger-sense", sense };
return sendAndReceive(socketPath, message, parseDaemonResponse);
}
/**
* Send a list-senses message to the running daemon via its Unix socket.
* Resolves with the list of registered senses or rejects on connection/timeout errors.
*/
export function listSensesViaDaemon(socketPath: string): Promise<ListSensesResponse> {
return sendAndReceive(socketPath, { type: "list-senses" }, parseListSensesResponse);
export function listSensesViaDaemon(socketPath: string): Promise<DaemonIpcListSensesResponse> {
const message: DaemonIpcRequest = { type: "list-senses" };
return sendAndReceive(socketPath, message, parseListSensesResponse);
}
@@ -0,0 +1,51 @@
import { describe, expect, it } from "vitest";
import { parseDaemonIpcRequest } from "../daemon-ipc-protocol.js";
describe("parseDaemonIpcRequest", () => {
it("parses trigger-workflow", () => {
expect(
parseDaemonIpcRequest(
JSON.stringify({
type: "trigger-workflow",
workflow: "wf",
prompt: "go",
maxRounds: 3,
}),
),
).toEqual({
type: "trigger-workflow",
workflow: "wf",
prompt: "go",
maxRounds: 3,
});
});
it("rejects trigger-workflow with empty workflow", () => {
expect(
parseDaemonIpcRequest(
JSON.stringify({
type: "trigger-workflow",
workflow: "",
prompt: "",
maxRounds: 1,
}),
),
).toBeNull();
});
it("parses trigger-sense and list-senses", () => {
expect(parseDaemonIpcRequest(JSON.stringify({ type: "trigger-sense", sense: "x" }))).toEqual({
type: "trigger-sense",
sense: "x",
});
expect(parseDaemonIpcRequest(JSON.stringify({ type: "list-senses" }))).toEqual({
type: "list-senses",
});
});
it("returns null for invalid JSON or unknown type", () => {
expect(parseDaemonIpcRequest("not json")).toBeNull();
expect(parseDaemonIpcRequest(JSON.stringify({ type: "nope" }))).toBeNull();
});
});
+85
View File
@@ -0,0 +1,85 @@
/**
* Daemon Unix-socket IPC protocol (CLI → daemon).
* Newline-delimited JSON: one request object per line from the client,
* one response object per line from the daemon.
*/
import { isPlainRecord } from "./is-plain-record.js";
import type { SenseInfo } from "./types.js";
/** Client → daemon: start a workflow run. */
export type DaemonIpcTriggerWorkflowRequest = {
type: "trigger-workflow";
workflow: string;
prompt: string;
maxRounds: number;
};
/** Client → daemon: run a sense compute on demand. */
export type DaemonIpcTriggerSenseRequest = {
type: "trigger-sense";
sense: string;
};
/** Client → daemon: list registered senses. */
export type DaemonIpcListSensesRequest = {
type: "list-senses";
};
/** Union of all JSON requests the daemon IPC server accepts. */
export type DaemonIpcRequest =
| DaemonIpcTriggerWorkflowRequest
| DaemonIpcTriggerSenseRequest
| DaemonIpcListSensesRequest;
/** Successful trigger / trigger-sense reply (no body). */
export type DaemonIpcTriggerOkResponse = { ok: true };
export type DaemonIpcErrorResponse = { ok: false; error: string };
/** Replies for trigger-workflow and trigger-sense. */
export type DaemonIpcTriggerResponse = DaemonIpcTriggerOkResponse | DaemonIpcErrorResponse;
/** Reply for list-senses. */
export type DaemonIpcListSensesResponse =
| { ok: true; senses: SenseInfo[] }
| DaemonIpcErrorResponse;
/** Any JSON response the daemon may write on the IPC socket. */
export type DaemonIpcResponse =
| DaemonIpcTriggerOkResponse
| DaemonIpcErrorResponse
| { ok: true; senses: SenseInfo[] };
/**
* Parse a single line of JSON into a {@link DaemonIpcRequest}, or null if invalid.
* Kept in core with the request types so CLI and daemon stay aligned at compile time.
*/
export function parseDaemonIpcRequest(line: string): DaemonIpcRequest | null {
try {
const obj: unknown = JSON.parse(line);
if (!isPlainRecord(obj)) return null;
const req = obj;
if (req.type === "trigger-workflow") {
if (typeof req.workflow !== "string" || req.workflow.length === 0) return null;
if (typeof req.prompt !== "string") return null;
if (typeof req.maxRounds !== "number") return null;
return {
type: "trigger-workflow",
workflow: req.workflow,
prompt: req.prompt,
maxRounds: req.maxRounds,
};
}
if (req.type === "trigger-sense") {
if (typeof req.sense !== "string" || req.sense.length === 0) return null;
return { type: "trigger-sense", sense: req.sense };
}
if (req.type === "list-senses") {
return { type: "list-senses" };
}
return null;
} catch {
return null;
}
}
+13
View File
@@ -32,3 +32,16 @@ export {
parseSenseWorkflowDirective,
routeSenseComputeOutput,
} from "./sense-workflow-directive.js";
export type {
DaemonIpcTriggerWorkflowRequest,
DaemonIpcTriggerSenseRequest,
DaemonIpcListSensesRequest,
DaemonIpcRequest,
DaemonIpcTriggerOkResponse,
DaemonIpcErrorResponse,
DaemonIpcTriggerResponse,
DaemonIpcListSensesResponse,
DaemonIpcResponse,
} from "./daemon-ipc-protocol.js";
export { parseDaemonIpcRequest } from "./daemon-ipc-protocol.js";
@@ -2,7 +2,7 @@
* Unit + integration tests for daemon-ipc.ts — trigger-sense request type.
*
* Tests cover:
* - parseRequest correctly accepts/rejects trigger-sense messages
* - parseDaemonIpcRequest (core) / server correctly accept or reject trigger-sense messages
* - createDaemonIpcServer routes trigger-sense to opts.triggerSense
* - Error response when triggerSense throws (unknown sense)
* - Success response on valid sense trigger
+13 -69
View File
@@ -2,83 +2,24 @@
* Daemon IPC server — listens on a Unix domain socket so that the CLI
* can send commands (e.g. trigger-workflow, trigger-sense) to the running daemon process.
*
* Protocol: newline-delimited JSON messages.
* Each request: { type: "trigger-workflow"; workflow: string; payload: unknown }
* | { type: "trigger-sense"; sense: string }
* | { type: "list-senses" }
* Each response: { ok: true } | { ok: false; error: string }
* | { ok: true; senses: SenseInfo[] } (for list-senses)
* Protocol: newline-delimited JSON — request/response types and
* `parseDaemonIpcRequest` live in `@uncaged/nerve-core`.
*/
import { rmSync } from "node:fs";
import { type Server, type Socket, createServer } from "node:net";
import type { SenseInfo } from "@uncaged/nerve-core";
import { isPlainRecord } from "@uncaged/nerve-core";
import type { DaemonIpcResponse, SenseInfo } from "@uncaged/nerve-core";
import { parseDaemonIpcRequest } from "@uncaged/nerve-core";
import type { WorkflowManager } from "./workflow-manager.js";
export type { SenseInfo };
/** JSON message sent by the CLI to trigger a workflow. */
export type TriggerWorkflowRequest = {
type: "trigger-workflow";
workflow: string;
prompt: string;
maxRounds: number;
};
/** JSON message sent by the CLI to trigger a sense compute on-demand. */
export type TriggerSenseRequest = {
type: "trigger-sense";
sense: string;
};
/** JSON message sent by the CLI to list registered senses. */
export type ListSensesRequest = {
type: "list-senses";
};
type DaemonRequest = TriggerWorkflowRequest | TriggerSenseRequest | ListSensesRequest;
type DaemonResponse =
| { ok: true }
| { ok: false; error: string }
| { ok: true; senses: SenseInfo[] };
export type DaemonIpcServer = {
close: () => Promise<void>;
};
function parseRequest(line: string): DaemonRequest | null {
try {
const obj: unknown = JSON.parse(line);
if (!isPlainRecord(obj)) return null;
const req = obj;
if (req.type === "trigger-workflow") {
if (typeof req.workflow !== "string" || req.workflow.length === 0) return null;
if (typeof req.prompt !== "string") return null;
if (typeof req.maxRounds !== "number") return null;
return {
type: "trigger-workflow",
workflow: req.workflow,
prompt: req.prompt,
maxRounds: req.maxRounds,
};
}
if (req.type === "trigger-sense") {
if (typeof req.sense !== "string" || req.sense.length === 0) return null;
return { type: "trigger-sense", sense: req.sense };
}
if (req.type === "list-senses") {
return { type: "list-senses" };
}
return null;
} catch {
return null;
}
}
export type DaemonIpcServerOptions = {
/** Called when a trigger-sense request arrives. Should throw if the sense is unknown. */
triggerSense: (senseName: string) => void;
@@ -102,9 +43,9 @@ export function createDaemonIpcServer(
const trimmed = line.trim();
if (trimmed.length === 0) return;
const req = parseRequest(trimmed);
const req = parseDaemonIpcRequest(trimmed);
if (req === null) {
const resp: DaemonResponse = { ok: false, error: "Invalid request" };
const resp: DaemonIpcResponse = { ok: false, error: "Invalid request" };
socket.write(`${JSON.stringify(resp)}\n`);
return;
}
@@ -115,20 +56,23 @@ export function createDaemonIpcServer(
prompt: req.prompt,
maxRounds: req.maxRounds,
});
const resp: DaemonResponse = { ok: true };
const resp: DaemonIpcResponse = { ok: true };
socket.write(`${JSON.stringify(resp)}\n`);
} else if (req.type === "trigger-sense") {
opts.triggerSense(req.sense);
const resp: DaemonResponse = { ok: true };
const resp: DaemonIpcResponse = { ok: true };
socket.write(`${JSON.stringify(resp)}\n`);
} else if (req.type === "list-senses") {
const senses = opts.listSenses();
const resp: DaemonResponse = { ok: true, senses };
const resp: DaemonIpcResponse = { ok: true, senses };
socket.write(`${JSON.stringify(resp)}\n`);
} else {
const _exhaustive: never = req;
void _exhaustive;
}
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
const resp: DaemonResponse = { ok: false, error: msg };
const resp: DaemonIpcResponse = { ok: false, error: msg };
socket.write(`${JSON.stringify(resp)}\n`);
}
}
+1 -1
View File
@@ -30,7 +30,7 @@ export {
export { createKernel } from "./kernel.js";
export type { Kernel, KernelOptions, KernelHealth } from "./kernel.js";
export type { SenseInfo } from "./daemon-ipc.js";
export type { SenseInfo } from "@uncaged/nerve-core";
export { createFileWatcher } from "./file-watcher.js";
export type { FileWatcher, FileChange, FileChangeHandler } from "./file-watcher.js";