fix: address review issues #46-#49 #52

Merged
xiaomo merged 1 commits from fix/review-issues-46-49 into main 2026-04-23 00:24:19 +00:00
11 changed files with 117 additions and 109 deletions
@@ -3,6 +3,7 @@
* If the daemon package changes its public API, this file will fail to compile.
*/
import type { SenseInfo } from "@uncaged/nerve-core";
import type {
ArchiveLogsDayResult as DaemonArchiveLogsDayResult,
ArchiveLogsOptions as DaemonArchiveLogsOptions,
@@ -10,6 +11,7 @@ import type {
LogEntry as DaemonLogEntry,
LogQuery as DaemonLogQuery,
LogStore as DaemonLogStore,
SenseInfo as DaemonSenseInfo,
WorkflowRun as DaemonWorkflowRun,
WorkflowRunStatus as DaemonWorkflowRunStatus,
} from "@uncaged/nerve-daemon";
@@ -27,6 +29,11 @@ import type {
} from "../daemon-types.js";
describe("daemon-types drift guard", () => {
it("SenseInfo matches daemon package export (list-senses IPC)", () => {
expectTypeOf<SenseInfo>().toMatchTypeOf<DaemonSenseInfo>();
expectTypeOf<DaemonSenseInfo>().toMatchTypeOf<SenseInfo>();
});
it("WorkflowRunStatus is assignable both ways", () => {
expectTypeOf<WorkflowRunStatus>().toMatchTypeOf<DaemonWorkflowRunStatus>();
expectTypeOf<DaemonWorkflowRunStatus>().toMatchTypeOf<WorkflowRunStatus>();
@@ -13,18 +13,24 @@ import { createServer } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { SenseInfo } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { listSensesViaDaemon } from "../daemon-client.js";
import type { SenseInfo } from "../daemon-client.js";
import { formatDuration, formatSenseList, sensesFromConfig } from "../commands/sense.js";
import { listSensesViaDaemon } from "../daemon-client.js";
// ---------------------------------------------------------------------------
// Fixtures
// ---------------------------------------------------------------------------
const SAMPLE_SENSES: SenseInfo[] = [
{ name: "cpu-usage", group: "system", throttle: 5000, timeout: 3000, lastSignalTs: 1_700_000_000_000 },
{
name: "cpu-usage",
group: "system",
throttle: 5000,
timeout: 3000,
lastSignalTs: 1_700_000_000_000,
},
{ name: "disk-usage", group: "system", throttle: 30000, timeout: null, lastSignalTs: null },
{ name: "active-tasks", group: "tasks", throttle: 10000, timeout: 30000, lastSignalTs: null },
];
+2 -4
View File
@@ -1,11 +1,10 @@
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { parseNerveConfig } from "@uncaged/nerve-core";
import { type SenseInfo, parseNerveConfig } from "@uncaged/nerve-core";
import { defineCommand } from "citty";
import { listSensesViaDaemon, triggerSenseViaDaemon } from "../daemon-client.js";
import type { SenseInfo } from "../daemon-client.js";
import { getNerveRoot, getSocketPath, isRunning } from "../workspace.js";
// ---------------------------------------------------------------------------
@@ -35,8 +34,7 @@ export function formatSenseList(senses: SenseInfo[]): string {
lines.push(` group: ${s.group}\n`);
lines.push(` throttle: ${formatDuration(s.throttle)}\n`);
lines.push(` timeout: ${formatDuration(s.timeout)}\n`);
const lastSignal =
s.lastSignalTs !== null ? new Date(s.lastSignalTs).toISOString() : "(never)";
const lastSignal = s.lastSignalTs !== null ? new Date(s.lastSignalTs).toISOString() : "(never)";
lines.push(` last signal: ${lastSignal}\n`);
}
return lines.join("");
+40 -85
View File
@@ -8,18 +8,14 @@
import { connect } from "node:net";
import type { Socket } from "node:net";
import type { SenseInfo } from "@uncaged/nerve-core";
const CONNECT_TIMEOUT_MS = 3_000;
const RESPONSE_TIMEOUT_MS = 5_000;
type TriggerResponse = { ok: true } | { ok: false; error: string };
export type { SenseInfo };
export type SenseInfo = {
name: string;
group: string;
throttle: number | null;
timeout: number | null;
lastSignalTs: number | null;
};
type TriggerResponse = { ok: true } | { ok: false; error: string };
type ListSensesResponse = { ok: true; senses: SenseInfo[] } | { ok: false; error: string };
@@ -37,12 +33,36 @@ function parseDaemonResponse(line: string): TriggerResponse {
return { ok: false, error: `Unexpected daemon response: ${line}` };
}
function sendAndReceive(socketPath: string, message: object): Promise<TriggerResponse> {
function parseListSensesResponse(line: string): ListSensesResponse {
try {
const obj = JSON.parse(line) as unknown;
if (obj !== null && typeof obj === "object") {
const r = obj as Record<string, unknown>;
if (r.ok === false && typeof r.error === "string") return { ok: false, error: r.error };
if (r.ok === true && Array.isArray(r.senses))
return { ok: true, senses: r.senses as SenseInfo[] };
}
} catch {
// fall through
}
return { ok: false, error: `Unexpected daemon response: ${line}` };
}
/**
* Connect to the daemon socket, send one JSON request (newline-terminated),
* and resolve with the first non-empty line parsed by `parseFirstLine`.
*/
function sendAndReceive<T>(
socketPath: string,
message: object,
parseFirstLine: (trimmed: string) => T,
responseTimeoutMs: number = RESPONSE_TIMEOUT_MS,
): Promise<T> {
return new Promise((resolve, reject) => {
let socket: Socket | null = null;
let settled = false;
function settle(result: TriggerResponse | Error): void {
function settle(result: T | Error): void {
if (settled) return;
settled = true;
if (socket !== null) {
@@ -65,7 +85,7 @@ function sendAndReceive(socketPath: string, message: object): Promise<TriggerRes
const responseTimer = setTimeout(() => {
settle(new Error("Timed out waiting for daemon response"));
}, RESPONSE_TIMEOUT_MS);
}, responseTimeoutMs);
let buf = "";
socket?.on("data", (chunk: Buffer) => {
@@ -76,7 +96,7 @@ function sendAndReceive(socketPath: string, message: object): Promise<TriggerRes
const trimmed = line.trim();
if (trimmed.length === 0) continue;
clearTimeout(responseTimer);
settle(parseDaemonResponse(trimmed));
settle(parseFirstLine(trimmed));
return;
}
});
@@ -101,18 +121,19 @@ export function triggerWorkflowViaDaemon(
workflow: string,
payload: unknown,
): Promise<TriggerResponse> {
return sendAndReceive(socketPath, { type: "trigger-workflow", workflow, payload });
return sendAndReceive(
socketPath,
{ type: "trigger-workflow", workflow, payload },
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 });
export function triggerSenseViaDaemon(socketPath: string, sense: string): Promise<TriggerResponse> {
return sendAndReceive(socketPath, { type: "trigger-sense", sense }, parseDaemonResponse);
}
/**
@@ -120,71 +141,5 @@ export function triggerSenseViaDaemon(
* Resolves with the list of registered senses or rejects on connection/timeout errors.
*/
export function listSensesViaDaemon(socketPath: string): Promise<ListSensesResponse> {
return new Promise((resolve, reject) => {
let socket: Socket | null = null;
let settled = false;
function settle(result: ListSensesResponse | Error): void {
if (settled) return;
settled = true;
if (socket !== null) {
socket.destroy();
socket = null;
}
if (result instanceof Error) {
reject(result);
} else {
resolve(result);
}
}
const connectTimer = setTimeout(() => {
settle(new Error(`Timed out connecting to daemon socket: ${socketPath}`));
}, CONNECT_TIMEOUT_MS);
socket = connect(socketPath, () => {
clearTimeout(connectTimer);
const responseTimer = setTimeout(() => {
settle(new Error("Timed out waiting for daemon response"));
}, RESPONSE_TIMEOUT_MS);
let buf = "";
socket?.on("data", (chunk: Buffer) => {
buf += chunk.toString("utf8");
const lines = buf.split("\n");
buf = lines.pop() ?? "";
for (const line of lines) {
const trimmed = line.trim();
if (trimmed.length === 0) continue;
clearTimeout(responseTimer);
try {
const obj = JSON.parse(trimmed) as unknown;
if (obj !== null && typeof obj === "object") {
const r = obj as Record<string, unknown>;
if (r.ok === false && typeof r.error === "string") {
settle({ ok: false, error: r.error });
return;
}
if (r.ok === true && Array.isArray(r.senses)) {
settle({ ok: true, senses: r.senses as SenseInfo[] });
return;
}
}
} catch {
// fall through
}
settle({ ok: false, error: `Unexpected daemon response: ${trimmed}` });
return;
}
});
socket?.write(`${JSON.stringify({ type: "list-senses" })}\n`);
});
socket.on("error", (err) => {
clearTimeout(connectTimer);
settle(new Error(`Cannot connect to daemon: ${err.message}`));
});
});
return sendAndReceive(socketPath, { type: "list-senses" }, parseListSensesResponse);
}
+1
View File
@@ -1,6 +1,7 @@
export type {
Signal,
SenseConfig,
SenseInfo,
SenseReflexConfig,
WorkflowReflexConfig,
ReflexConfig,
+9
View File
@@ -12,6 +12,15 @@ export type SenseConfig = {
gracePeriod: number | null;
};
/** Runtime metadata for a sense (e.g. daemon list-senses IPC). */
export type SenseInfo = {
name: string;
group: string;
throttle: number | null;
timeout: number | null;
lastSignalTs: number | null;
};
export type SenseReflexConfig = {
kind: "sense";
sense: string;
@@ -1,4 +1,7 @@
import { EventEmitter } from "node:events";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { NerveConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@@ -44,6 +47,7 @@ vi.mock("node:child_process", () => ({
// Import after mock is set up
const { createKernel } = await import("../kernel.js");
const { createLogStore } = await import("../log-store.js");
// ---------------------------------------------------------------------------
// Helpers
@@ -93,6 +97,29 @@ describe("kernel — message routing", () => {
await kernel.stop();
});
it("persists emitted signals as sense/signal log entries", async () => {
const tmpDir = mkdtempSync(join(tmpdir(), "nerve-kernel-sig-"));
const logStore = createLogStore(join(tmpDir, "logs.db"));
try {
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [],
});
const kernel = createKernel(config, tmpDir, { logStore });
const child = mockChildren[0];
child.emit("message", { type: "ready" });
child.emit("message", { type: "signal", sense: "cpu-usage", payload: 123 });
const rows = logStore.query({ source: "sense", type: "signal", refId: "cpu-usage" });
expect(rows).toHaveLength(1);
expect(rows[0].payload).toBe(JSON.stringify(123));
await kernel.stop();
} finally {
rmSync(tmpDir, { recursive: true, force: true });
}
});
it("routes error message to stderr", async () => {
const stderrSpy = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
const config = makeConfig({
+4 -9
View File
@@ -13,8 +13,12 @@
import { rmSync } from "node:fs";
import { type Server, type Socket, createServer } from "node:net";
import type { SenseInfo } 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";
@@ -33,15 +37,6 @@ export type ListSensesRequest = {
type: "list-senses";
};
/** Runtime info about a single sense returned by list-senses. */
export type SenseInfo = {
name: string;
group: string;
throttle: number | null;
timeout: number | null;
lastSignalTs: number | null;
};
type DaemonRequest = TriggerWorkflowRequest | TriggerSenseRequest | ListSensesRequest;
type DaemonResponse =
+2
View File
@@ -29,6 +29,8 @@ export {
export { createKernel } from "./kernel.js";
export type { Kernel, KernelOptions, KernelHealth } from "./kernel.js";
export type { SenseInfo } from "./daemon-ipc.js";
export { createFileWatcher } from "./file-watcher.js";
export type { FileWatcher, FileChange, FileChangeHandler } from "./file-watcher.js";
+11 -7
View File
@@ -18,11 +18,11 @@ import { readFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import type { NerveConfig, Signal } from "@uncaged/nerve-core";
import type { NerveConfig, SenseInfo, Signal } from "@uncaged/nerve-core";
import { parseNerveConfig } from "@uncaged/nerve-core";
import { createDaemonIpcServer } from "./daemon-ipc.js";
import type { DaemonIpcServer, SenseInfo } from "./daemon-ipc.js";
import type { DaemonIpcServer } from "./daemon-ipc.js";
import { createFileWatcher } from "./file-watcher.js";
import type { FileWatcher } from "./file-watcher.js";
import type { ComputeMessage, ShutdownMessage } from "./ipc.js";
@@ -89,7 +89,11 @@ function spawnWorker(nerveRoot: string, group: string, workerScript: string): Ch
stdio: ["ignore", "inherit", "inherit", "ipc"],
});
// Prevent unhandled EPIPE when writing to a child whose IPC channel closed
child.on("error", () => {});
child.on("error", (err) => {
if ((err as NodeJS.ErrnoException).code !== "EPIPE") {
console.error("[worker] error:", err.message);
}
});
return child;
}
@@ -222,8 +226,8 @@ export function createKernel(
ts: Date.now(),
};
logStore.append({
source: "reflex",
type: "run_complete",
source: "sense",
type: "signal",
refId: msg.sense,
payload: JSON.stringify(msg.payload),
ts: signal.ts,
@@ -524,8 +528,8 @@ export function createKernel(
listSenses(): SenseInfo[] {
return Object.entries(config.senses).map(([name, senseConfig]) => {
const entries = logStore.query({
source: "reflex",
type: "run_complete",
source: "sense",
type: "signal",
refId: name,
});
const lastEntry = entries.length > 0 ? entries[entries.length - 1] : null;
+5 -1
View File
@@ -90,7 +90,11 @@ function spawnWorkflowWorker(
stdio: ["ignore", "inherit", "inherit", "ipc"],
});
// Prevent unhandled EPIPE when writing to a child whose IPC channel closed
child.on("error", () => {});
child.on("error", (err) => {
if ((err as NodeJS.ErrnoException).code !== "EPIPE") {
console.error("[worker] error:", err.message);
}
});
return child;
}