Compare commits

..

53 Commits

Author SHA1 Message Date
xiaoju 640f170de8 refactor: add daemon subcommand group and dev foreground mode
- Create 'nerve daemon' subcommand group: start, stop, status, restart, logs
- Create 'nerve dev' for foreground mode (replaces old start without -d)
- 'nerve daemon start' is always background (removed -d/--daemon flag)
- Keep top-level aliases: nerve start/stop/status/logs → nerve daemon *
- Extract runStopCommand() for restart reuse
- Add daemon-cli tests

Closes #53

小橘 🍊(NEKO Team)
2026-04-23 01:16:13 +00:00
xiaoju 119b1f3722 chore: enforce pnpm publish for all packages unconditionally
小橘 <xiaoju@shazhou.work>
2026-04-23 00:49:39 +00:00
xiaoju 96ea4b46ff chore: add prepublish guard against npm publish with workspace:* deps
小橘 <xiaoju@shazhou.work>
2026-04-23 00:47:56 +00:00
xiaoju 57881533a8 docs: fix publish skill — use pnpm publish for workspace:* conversion
小橘 <xiaoju@shazhou.work>
2026-04-23 00:43:51 +00:00
xiaoju a62a993a82 fix(cli): remove duplicate shebang in daemon-bootstrap causing crash on nerve start -d
小橘 <xiaoju@shazhou.work>
2026-04-23 00:43:18 +00:00
xiaoju 3f22eb4664 release: @uncaged/nerve-core@0.1.3, @uncaged/nerve-daemon@0.1.4, @uncaged/nerve-cli@0.1.5
小橘 <xiaoju@shazhou.work>
2026-04-23 00:35:40 +00:00
xiaoju b5913263e4 docs: add publish and setup skills
小橘 <xiaoju@shazhou.work>
2026-04-23 00:31:27 +00:00
xiaomo d3ecd2a492 Merge pull request 'fix: address review issues #46-#49' (#52) from fix/review-issues-46-49 into main 2026-04-23 00:24:19 +00:00
xiaoju 8763440436 fix: address review issues #46-#49
#46 — EPIPE handler: only silence EPIPE, log other child errors
#47 — lastSignalTs: query sense/signal instead of reflex/run_complete
#48 — SenseInfo: deduplicate to @uncaged/nerve-core, add expectTypeOf test
#49 — IPC client: extract sendAndReceive<T> to eliminate duplication

小橘 <xiaoju@shazhou.work>
2026-04-23 00:22:55 +00:00
xiaomo f270804002 Merge pull request 'feat(daemon): CAS blob store — sha256 content-addressable storage (closes #39)' (#51) from feat/blob-store into main 2026-04-23 00:21:46 +00:00
xiaoju 404ee3e34f feat(daemon): add CAS blob store with sha256 content-addressable storage — closes #39
- createBlobStore(root) with write/read/exists API
- sha256 hex, first 2 chars as shard directory
- Atomic writes via temp file + rename
- CAS mismatch detection on read and write
- Inject blobStore into sense compute via options.blobs
- Export createBlobStore, normalizeBlobHash, BlobStore type
2026-04-23 00:19:35 +00:00
xiaomo cbc6db6b7d Merge pull request 'feat(daemon): log store archival — Meta table + JSONL cold archive (closes #38)' (#45) from feat/log-archive into main 2026-04-23 00:17:54 +00:00
xiaomo b1f6c775ce Merge pull request 'fix(init): auto-verify and retry better-sqlite3 native build — closes #44' (#50) from fix/init-sqlite-retry into main 2026-04-23 00:14:30 +00:00
xingyue 4ada5ef335 fix(init): auto-verify and retry better-sqlite3 native build
After pnpm install, verify better-sqlite3 actually loads by spawning
a test process. If it fails, rebuild up to 2 times. On final failure,
print actionable fix commands instead of a vague warning.

Closes #44
2026-04-23 08:12:10 +08:00
xiaoju 978b1680a3 feat(daemon): add log store archival with meta watermark + JSONL cold archive — closes #38
- Add meta table with archived_up_to watermark in logs.db
- Archive logs older than 30 days to data/archive/logs/YYYY-MM-DD.jsonl
- Idempotent: same-day re-export overwrites file
- Single transaction: DELETE + UPDATE meta
- Optional VACUUM after archive loop
- CLI: nerve store archive [--vacuum]
- 15+ new tests for archive logic
2026-04-23 00:10:20 +00:00
xiaoju ac34b798c2 feat(cli): add nerve sense list command with IPC + static fallback — closes #37
- daemon-ipc: add list-senses request type returning SenseInfo[]
- kernel: implement listSenses querying logStore for last signal time
- CLI: nerve sense list with table output, fallback to nerve.yaml when daemon is down
- 25 new tests across daemon-ipc and CLI
2026-04-23 00:00:23 +00:00
xiaoju 00c9b7e406 test: add trigger-sense unit + integration tests — closes #36
- daemon-ipc: parse trigger-sense request, success/failure responses
- kernel: triggerSense routing to correct worker, unknown sense error
- CLI: triggerSenseViaDaemon IPC round-trip
2026-04-22 23:53:23 +00:00
xiaoju 8b216e3f01 Revert "feat(cli): add nerve init sense <name> scaffold command — closes #36"
This reverts commit 7ded3a758a.
2026-04-22 23:44:18 +00:00
xiaoju 7ded3a758a feat(cli): add nerve init sense <name> scaffold command — closes #36
Implements nerve init sense <name> command that scaffolds a new sense directory under ~/.uncaged-nerve/senses/<name>/ with schema.ts, index.js, and migrations/0001_init.sql. Also auto-patches nerve.yaml to add the sense config and reflex entry. Includes full test coverage for all exported helpers.

Made-with: Cursor
2026-04-22 23:43:30 +00:00
xiaoju 3257237ba7 fix: handle EPIPE on child process IPC during shutdown
Add error event listener to forked workers in kernel and
workflow-manager to prevent unhandled EPIPE crashes.

Closes #43
小橘 <xiaoju@shazhou.work>
2026-04-22 23:36:48 +00:00
xiaoju 2be11ac81a chore: release core@0.1.2 daemon@0.1.2 cli@0.1.3
小橘 <xiaoju@shazhou.work>
2026-04-22 23:12:29 +00:00
xiaomo 5ed4dfdde3 Merge pull request 'refactor(cli): decouple daemon native deps from CLI global install — closes #41' (#42) from refactor/decouple-daemon-from-cli into main 2026-04-22 23:09:56 +00:00
xingyue 282a802f06 fix: address review feedback on PR #42
1. [BLOCKER] tsup.config.ts: resolve merge conflict — keep both banner
   (shebang) and external (daemon decoupling)

2. [SHOULD-FIX] assertWorkspaceDaemonInstalled: throw Error instead of
   process.exit(1) — callers decide error handling

3. [SHOULD-FIX] getDaemonEntryPath: read daemon's package.json 'main'
   field instead of hardcoding dist/index.js

4. [SHOULD-FIX] daemon startup check: replace sleep(1500) with IPC
   socket polling (200ms intervals, 5s timeout)

5. [SHOULD-FIX] daemon-types drift: add vitest type-level assertions
   that verify CLI mirror types stay assignable with daemon exports
2026-04-23 07:07:38 +08:00
xingyue c8e6409837 refactor(cli): decouple daemon native deps from CLI global install
- Move @uncaged/nerve-daemon from runtime to devDependencies
- Dynamic import daemon from workspace node_modules at runtime
- Add daemon-bootstrap.ts as separate entry for background daemon spawn
- Extract run-foreground-kernel.ts and workspace-daemon.ts modules
- Add daemon-types.ts for structural types (no runtime daemon import)
- Rebuild better-sqlite3 in workspace during nerve init
- Validate daemon process liveness after spawn in background mode
- Mark @uncaged/nerve-daemon as external in tsup config

Closes #41
2026-04-23 06:58:00 +08:00
xiaoju 877da470d7 fix: pre-approve build scripts in nerve init scaffold
Add pnpm.onlyBuiltDependencies for better-sqlite3 and esbuild
to suppress pnpm v10 approve-builds warnings.

小橘 <xiaoju@shazhou.work>
2026-04-22 15:49:13 +00:00
xiaoju 01f54d14c5 chore: bump to 0.1.1 for npm publish fix
小橘 <xiaoju@shazhou.work>
2026-04-22 15:37:30 +00:00
xiaoju 6a689c4094 feat: make nerve-cli and nerve-daemon publishable npm packages
- Remove private:true from cli and daemon package.json
- Add files and publishConfig fields
- Add shebang banner via tsup for CLI entry
- Add trigger-sense IPC support in daemon and client

Closes #40

小橘 <xiaoju@shazhou.work>
2026-04-22 15:28:05 +00:00
xiaomo e66a376a77 Merge pull request 'feat: add nerve logs command with AI-friendly pagination — closes #29' (#34) from feat/nerve-logs into main 2026-04-22 15:04:52 +00:00
xiaoju 10f942b577 fix: address PR #34 review — SIGINT leak, negative offset, follow race conditions
- SIGINT: use process.once instead of process.on
- Negative offset: validate and exit(1) with error to stderr
- Follow mode: sequential while loop replaces setInterval (no async race)
- Log rotation: reset size when newSize < size
- TODO: readAllLines large file optimization note
- 2 new tests for negative offset validation

小橘 <xiaoju@shazhou.work>
2026-04-22 15:00:24 +00:00
xiaoju 76b547d37a feat: add nerve logs command with AI-friendly pagination — closes #29
- nerve logs: tail last 50 lines by default
- -n <lines>: specify line count
- --offset <n>: pagination from line n (1-based)
- -f/--follow: real-time tail with 300ms polling
- Footer with stats + next-page command hint for AI agents
- No ANSI colors, emoji only, data→stdout, errors→stderr
- 19 new tests covering pagination, footer, edge cases

小橘 <xiaoju@shazhou.work>
2026-04-22 14:52:17 +00:00
xiaoju 1b2ff37097 chore: publish @uncaged/nerve-core@0.0.1 to npm — closes #28
Removed 'private: true' to allow npm publish. Package is now available
at https://www.npmjs.com/package/@uncaged/nerve-core

小橘 <xiaoju@shazhou.work>
2026-04-22 14:37:07 +00:00
xiaoju 4add0d88c6 Revert "Merge pull request 'fix: remove unpublished @uncaged/nerve-core from init template — closes #28' (#33) from fix/remove-unpublished-dep into main"
This reverts commit a8404dc096, reversing
changes made to 569c034b49.
2026-04-22 14:36:24 +00:00
xiaoju a8404dc096 Merge pull request 'fix: remove unpublished @uncaged/nerve-core from init template — closes #28' (#33) from fix/remove-unpublished-dep into main 2026-04-22 14:35:24 +00:00
xiaoju 891db36152 fix: remove unpublished @uncaged/nerve-core from init template — closes #28
The workspace package.json template listed @uncaged/nerve-core as a
dependency, but this package is not published to npm. Since the generated
workflow code only imports from @uncaged/nerve-daemon (which is also not
yet published but will be), remove the unnecessary dependency to unblock
`nerve init`.

小橘 <xiaoju@shazhou.work>
2026-04-22 14:35:03 +00:00
xiaoju 569c034b49 Merge pull request 'fix: daemon mode spawn path — closes #27' (#30) from fix/daemon-spawn-path into main 2026-04-22 14:21:33 +00:00
xingyue 85fa282d2e fix(cli): create initial git commit after workspace init
git init without add+commit leaves the workspace in a dirty state
with no baseline to diff against.
2026-04-22 22:16:41 +08:00
xiaomo b75a112c95 Merge pull request 'fix: IPC trigger try/catch + test import cleanup' (#32) from fix/phase4-followup into main 2026-04-22 14:16:10 +00:00
xingyue 606eff6d70 fix(cli): remove self-fallback in cliEntryScript candidates
Per review: third candidate (here) is wrong — if bundled and source
candidates both miss, falling back to self reproduces the original bug.
Keep only the two valid candidates and throw on miss.
2026-04-22 22:15:53 +08:00
xingyue 97305bd9af fix(cli): resolve CLI entry path for bundled dist output
cliEntryScript() assumed source directory structure (src/commands/start.ts → ../cli.ts),
but after tsup bundles everything into dist/cli.js, import.meta.url points to dist/cli.js
and the '../cli.js' path resolves to a non-existent file.

Use candidate-based lookup: try same-dir, parent-dir, then self (bundled case).
2026-04-22 22:15:53 +08:00
xingyue 3f2c9df75d refactor: simplify cliEntryScript() — remove multi-level fallback
Per review feedback from xiaoju: the three-level fallback was over-defensive.
Since start.ts and cli.ts have a fixed relative position (commands/start.ts → ../cli.ts),
we can derive the path directly from import.meta.url with an existsSync guard.

This makes path errors explicit (throw) instead of silently falling back to
a potentially wrong path.
2026-04-22 22:15:53 +08:00
xingyue 1511cfd595 fix: daemon spawn uses CLI entry path instead of command module
The runDaemon function was using import.meta.url (pointing to start.js)
as the script for the spawned child process. This meant the child ran
`node start.js start` which has no CLI entry logic and exits immediately.

Added cliEntryScript() that resolves to the correct CLI entry (cli.js)
regardless of whether the code is bundled or split into separate files.

Closes #27
2026-04-22 22:15:53 +08:00
xiaoju 362dc94582 fix: add try/catch to IPC trigger handler & import real buildWorkflowTemplate in test
- daemon-ipc: wrap startWorkflow() in try/catch so errors are sent back
  as {ok:false, error:msg} instead of silently dropping the socket
- init-workflow.test: import buildWorkflowTemplate from init.ts instead
  of maintaining an inline copy

Addresses review follow-up suggestions from PR #31.

小橘 <xiaoju@shazhou.work>
2026-04-22 14:15:19 +00:00
xiaomo 9e7de3b4e0 Merge pull request 'feat: Workflow Engine Phase 4 — CLI & User Experience' (#31) from feat/workflow-engine-phase4 into main 2026-04-22 14:12:14 +00:00
xiaoju 7320761277 fix(cli): address PR #31 review — 6 issues fixed
Critical:
1. trigger: use Unix socket IPC to daemon instead of direct DB write
   - new daemon-ipc.ts (server) + daemon-client.ts (client)
   - kernel accepts ipcSocketPath, auto-starts IPC server
2. init workflow: validate name (lowercase alphanumeric + hyphens only)

Should fix:
3. getAllWorkflowRuns: SQL query on workflow_runs table instead of O(n) scan
4. limit/offset: robust parseIntArg() helper with NaN handling
5. statusIcon: exhaustive switch with never type check
6. trigger: end-to-end Unix socket tests added

12 new tests. All 224 tests pass.

小橘 🍊(NEKO Team)
2026-04-22 14:10:43 +00:00
xiaoju 262c77175f feat(cli): Phase 4 — workflow CLI commands & scaffold
- workflow list: active runs with --all/--workflow/--limit/--offset pagination
- workflow inspect <runId>: thread details with event pagination
- workflow trigger <name>: manual trigger, outputs runId
- init workflow <name>: scaffold template under workflows/

AI-friendly design: no ANSI colors, emoji for readability, pagination
with stats + next-page hint on all list commands.

47 new tests (39 workflow + 8 init-workflow). All 212 tests pass.

小橘 🍊(NEKO Team)
2026-04-22 13:46:05 +00:00
xiaomo ae80aef6b4 Merge pull request 'feat: Workflow Engine Phase 3 — Crash Recovery, Hot Reload & Incremental Config' (#22) from feat/workflow-engine-phase3 into main 2026-04-22 13:26:29 +00:00
xiaoju 8d92928951 fix(daemon): address PR #22 review — 6 issues fixed
Critical:
1. replayAndResume: remove double moderate() call, reuse loop result
2. drainAndRespawn: check workflow still in config before respawn
3. drain: mark in-flight runs as 'interrupted' in DB before clearing

Should fix:
4. crash recovery: dedup runId before re-queuing/re-activating
5. drain timeout: DEFAULT_DRAIN_TIMEOUT_MS > WORKER_SHUTDOWN_TIMEOUT_MS
6. crash-loop protection: max 5 crashes in 60s window, then stop respawn

5 new tests added. All 173 tests pass.

小橘 🍊(NEKO Team)
2026-04-22 13:25:35 +00:00
xiaoju 49ed65a330 feat(daemon): Phase 3 — crash recovery, hot reload & incremental config
- workflow-manager: crash detection, worker respawn, thread resume from
  persisted events, drainAndRespawn() for hot reload
- log-store: getTriggerPayload(), getThreadEvents() for crash recovery
- file-watcher: detect workflow .ts file changes under workflows/
- kernel: handleWorkflowFileChange(), incremental workflow config updates
  on reloadConfig() (add/remove/update concurrency)
- ipc: resume-thread message type for crash recovery
- workflow-worker: handle resume-thread, rebuild ThreadState from events

28 new tests across 4 test files. All 168 tests pass.

小橘 🍊(NEKO Team)
2026-04-22 13:25:35 +00:00
xiaomo b7dfe42a96 Merge pull request 'fix: init runtime bugs - missing dir, .ts/.js mismatch, TS annotations' (#26) from fix/init-runtime-bugs into main 2026-04-22 13:22:52 +00:00
xingyue a887fc04ca fix: init creates data/senses dir, generates .js templates without TS annotations
- Add mkdirSync for data/senses/ in init command (#23)
- Add defensive mkdirSync in sense-runtime before DB open (#23)
- Change init template output from index.ts to index.js (#24)
- Remove TypeScript type annotations from CPU usage template (#25)

Closes #23, closes #24, closes #25
2026-04-22 21:15:42 +08:00
xiaomo d5931a9e19 Merge pull request 'feat: Workflow Engine Phase 2 — Kernel Integration' (#21) from feat/workflow-engine-phase2 into main 2026-04-22 12:45:43 +00:00
xiaoju f12f187d8d feat(daemon): Phase 2 — kernel ↔ workflow integration
- kernel.ts: create WorkflowManager, expose on Kernel, wire into
  ReflexScheduler via workflowTriggerFn, add activeWorkflows to
  KernelHealth, graceful shutdown ordering
- reflex-scheduler.ts: handle kind:'workflow' reflexes — subscribe to
  bus signals and delegate to workflowTriggerFn
- workflow-manager.ts: add totalActiveCount(), updateConfig() for hot
  reload support

All 140 tests pass.

小橘 🍊(NEKO Team)
2026-04-22 12:40:57 +00:00
xiaomo 1512113a01 Merge pull request 'feat: Workflow Engine Phase 1' (#17) from feat/workflow-engine-phase1 into main 2026-04-22 12:20:12 +00:00
60 changed files with 6659 additions and 336 deletions
+80
View File
@@ -0,0 +1,80 @@
# Skill: Publish @uncaged/nerve packages to npm
## When to use
When releasing a new version of any `@uncaged/nerve-*` package to npm.
## Prerequisites
- npm login with an account that has **owner** access to the `@uncaged` org
- All tests pass: `pnpm -r run test`
- Clean working tree (no uncommitted changes)
## Packages
| Package | Path | npm |
|---------|------|-----|
| `@uncaged/nerve-core` | `packages/core` | [link](https://www.npmjs.com/package/@uncaged/nerve-core) |
| `@uncaged/nerve-daemon` | `packages/daemon` | [link](https://www.npmjs.com/package/@uncaged/nerve-daemon) |
| `@uncaged/nerve-cli` | `packages/cli` | [link](https://www.npmjs.com/package/@uncaged/nerve-cli) |
## Dependency order
`core``daemon``cli`
Always publish in this order. If `core` has changes, bump and publish it first, then update dependents.
## Steps
### 1. Ensure clean state
```bash
git checkout main && git pull origin main
pnpm install
pnpm -r run build
pnpm -r run test
```
### 2. Bump versions
Manually update `version` in each changed package's `package.json`.
Follow semver:
- **patch** (0.1.x): bug fixes, refactors
- **minor** (0.x.0): new features, non-breaking API additions
- **major** (x.0.0): breaking changes
If bumping `core`, also update the `@uncaged/nerve-core` dependency version in `daemon` and `cli` package.json. Same for `daemon``cli`.
### 3. Build
```bash
pnpm -r run build
```
### 4. Publish (in order)
```bash
# Only publish packages that have version bumps
# MUST use pnpm publish (not npm) — pnpm converts workspace:* to real versions
cd packages/core && pnpm publish --access public --no-git-checks
cd packages/daemon && pnpm publish --access public --no-git-checks
cd packages/cli && pnpm publish --access public --no-git-checks
```
### 5. Commit & tag
```bash
git add -A
git commit -m "release: @uncaged/nerve-core@X.Y.Z, @uncaged/nerve-daemon@X.Y.Z, @uncaged/nerve-cli@X.Y.Z"
git tag -a vX.Y.Z -m "Release vX.Y.Z"
git push origin main --tags
```
## Pitfalls
- **Don't publish without building first** — `tsup` output in `dist/` is what npm ships
- **Dependency order matters** — if you publish `daemon` before `core`, npm may resolve the old `core` version
- **`--access public`** is required for scoped packages on first publish; safe to always include
- **Check `npm whoami`** to confirm you're logged in as the right account
- **No changeset tool** — this project uses manual version bumps (no changesets/lerna)
+101
View File
@@ -0,0 +1,101 @@
# Skill: Setup nerve from scratch
## When to use
Setting up the nerve project for local development from a fresh clone.
## Prerequisites
- **Node.js** ≥ 18
- **pnpm** ≥ 9 (`npm install -g pnpm`)
- **Git** access to `git.shazhou.work`
## Steps
### 1. Clone
```bash
git clone https://git.shazhou.work/uncaged/nerve.git
cd nerve
```
### 2. Install dependencies
```bash
pnpm install
```
This installs all workspace packages and links internal dependencies (`core``daemon``cli`).
### 3. Build all packages
```bash
pnpm -r run build
```
Build order is handled automatically by pnpm workspace — `core` builds first, then `daemon`, then `cli`.
### 4. Run tests
```bash
pnpm -r run test
```
Or test individual packages:
```bash
pnpm --filter @uncaged/nerve-core test
pnpm --filter @uncaged/nerve-daemon test
pnpm --filter @uncaged/nerve-cli test
```
### 5. Try the CLI
```bash
# Link the CLI globally
cd packages/cli && npm link
# Initialize a workspace
mkdir ~/my-nerve-workspace && cd ~/my-nerve-workspace
nerve init
# Edit senses in nerve.yaml, then:
nerve start # start the daemon
nerve sense list # list registered senses
nerve stop # stop the daemon
```
### 6. Lint & format
```bash
pnpm run check # biome lint check
pnpm run format # biome auto-format
```
## Project structure
```
nerve/
├── packages/
│ ├── core/ # @uncaged/nerve-core — shared types, log store, blob store
│ ├── daemon/ # @uncaged/nerve-daemon — kernel, sense runtime, workflow manager
│ └── cli/ # @uncaged/nerve-cli — CLI commands (init, start, stop, sense, etc.)
├── docs/ # RFCs, conventions, skills
├── pnpm-workspace.yaml
└── biome.json # linter/formatter config
```
## Key conventions
- **Monorepo** with pnpm workspaces
- **ESM only** — all packages output ESM (`"type": "module"`)
- **tsup** for builds, **vitest** for tests, **biome** for lint/format
- **SQLite** (better-sqlite3) for log store and blob store
- See `docs/coding-conventions.md` for code style rules
## Pitfalls
- **Must build before test** — daemon and cli import compiled output from core
- **better-sqlite3** requires native compilation — if `pnpm install` fails, ensure you have build tools (`build-essential` on Linux, Xcode CLI tools on macOS)
- **Node 18+** required — uses native `fetch`, `crypto.randomUUID`, etc.
- **pnpm only** — don't use npm/yarn, workspace links won't resolve correctly
+14 -5
View File
@@ -1,22 +1,31 @@
{
"name": "@uncaged/nerve-cli",
"version": "0.0.1",
"private": true,
"version": "0.1.7",
"type": "module",
"bin": {
"nerve": "dist/cli.js"
},
"main": "dist/index.js",
"types": "dist/index.d.ts",
"files": [
"dist"
],
"publishConfig": {
"access": "public"
},
"scripts": {
"build": "tsup"
"prepublishOnly": "bash ../../scripts/prepublish-check.sh",
"build": "tsup",
"test": "vitest run"
},
"dependencies": {
"@uncaged/nerve-core": "workspace:*",
"@uncaged/nerve-daemon": "workspace:*",
"citty": "^0.1.6"
},
"devDependencies": {
"@types/node": "^22.0.0"
"@uncaged/nerve-daemon": "workspace:*",
"@types/better-sqlite3": "^7.6.13",
"@types/node": "^22.0.0",
"vitest": "^4.1.5"
}
}
@@ -0,0 +1,28 @@
import { describe, expect, it } from "vitest";
import { daemonCommand } from "../commands/daemon.js";
import { devCommand } from "../commands/dev.js";
import { daemonStartCommand } from "../commands/start.js";
describe("nerve daemon command group", () => {
it("exposes start, stop, status, restart, and logs subcommands", () => {
const subs = daemonCommand.subCommands;
expect(subs).toBeDefined();
if (!subs) {
throw new Error("expected daemonCommand.subCommands");
}
expect(Object.keys(subs).sort()).toEqual(["logs", "restart", "start", "status", "stop"]);
});
it("shares the same start command object as top-level nerve start alias", () => {
const subs = daemonCommand.subCommands;
expect(subs?.start).toBe(daemonStartCommand);
});
});
describe("nerve dev", () => {
it("is a foreground dev command", () => {
expect(devCommand.meta?.name).toBe("dev");
expect(devCommand.meta?.description).toMatch(/foreground/i);
});
});
@@ -0,0 +1,80 @@
/**
* Compile-time check: daemon-types.ts stays in sync with @uncaged/nerve-daemon exports.
* 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,
ArchiveLogsResult as DaemonArchiveLogsResult,
LogEntry as DaemonLogEntry,
LogQuery as DaemonLogQuery,
LogStore as DaemonLogStore,
SenseInfo as DaemonSenseInfo,
WorkflowRun as DaemonWorkflowRun,
WorkflowRunStatus as DaemonWorkflowRunStatus,
} from "@uncaged/nerve-daemon";
import { describe, expectTypeOf, it } from "vitest";
import type {
ArchiveLogsDayResult,
ArchiveLogsOptions,
ArchiveLogsResult,
LogEntry,
LogQuery,
LogStore,
WorkflowRun,
WorkflowRunStatus,
} 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>();
});
it("WorkflowRun is assignable both ways", () => {
expectTypeOf<WorkflowRun>().toMatchTypeOf<DaemonWorkflowRun>();
expectTypeOf<DaemonWorkflowRun>().toMatchTypeOf<WorkflowRun>();
});
it("LogEntry is assignable both ways", () => {
expectTypeOf<LogEntry>().toMatchTypeOf<DaemonLogEntry>();
expectTypeOf<DaemonLogEntry>().toMatchTypeOf<LogEntry>();
});
it("LogQuery is assignable both ways", () => {
expectTypeOf<LogQuery>().toMatchTypeOf<DaemonLogQuery>();
expectTypeOf<DaemonLogQuery>().toMatchTypeOf<LogQuery>();
});
it("LogStore has all required methods", () => {
expectTypeOf<LogStore>().toMatchTypeOf<
Pick<
DaemonLogStore,
| "query"
| "getWorkflowRun"
| "getActiveWorkflowRuns"
| "getAllWorkflowRuns"
| "upsertWorkflowRun"
| "archiveLogs"
| "close"
>
>();
});
it("ArchiveLogs types match daemon", () => {
expectTypeOf<ArchiveLogsOptions>().toMatchTypeOf<DaemonArchiveLogsOptions>();
expectTypeOf<DaemonArchiveLogsOptions>().toMatchTypeOf<ArchiveLogsOptions>();
expectTypeOf<ArchiveLogsResult>().toMatchTypeOf<DaemonArchiveLogsResult>();
expectTypeOf<DaemonArchiveLogsResult>().toMatchTypeOf<ArchiveLogsResult>();
expectTypeOf<ArchiveLogsDayResult>().toMatchTypeOf<DaemonArchiveLogsDayResult>();
expectTypeOf<DaemonArchiveLogsDayResult>().toMatchTypeOf<ArchiveLogsDayResult>();
});
});
@@ -0,0 +1,81 @@
/**
* Tests for nerve init workflow scaffold logic.
*
* We test the file-generation path by isolating the template rendering,
* not by invoking the full citty command (which calls process.exit).
*/
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { buildWorkflowTemplate } from "../commands/init.js";
let tmpDir: string;
beforeEach(() => {
tmpDir = mkdtempSync(join(tmpdir(), "nerve-cli-init-test-"));
});
afterEach(() => {
rmSync(tmpDir, { recursive: true, force: true });
});
describe("buildWorkflowTemplate", () => {
it("includes the workflow name in the template", () => {
const tpl = buildWorkflowTemplate("my-workflow");
expect(tpl).toContain("my-workflow started");
});
it("contains WorkflowDefinition type import", () => {
const tpl = buildWorkflowTemplate("test");
expect(tpl).toContain("WorkflowDefinition");
expect(tpl).toContain("@uncaged/nerve-daemon");
});
it("contains a moderate function that returns null to signal completion", () => {
const tpl = buildWorkflowTemplate("test");
expect(tpl).toContain("return null");
expect(tpl).toContain("moderate");
});
it("contains a roles map with main role", () => {
const tpl = buildWorkflowTemplate("test");
expect(tpl).toContain("roles:");
expect(tpl).toContain("main:");
});
it("uses different names per call", () => {
const a = buildWorkflowTemplate("workflow-a");
const b = buildWorkflowTemplate("workflow-b");
expect(a).toContain("workflow-a started");
expect(b).toContain("workflow-b started");
expect(a).not.toContain("workflow-b");
});
it("produces valid TypeScript syntax (no unclosed braces)", () => {
const tpl = buildWorkflowTemplate("test");
const opens = (tpl.match(/\{/g) ?? []).length;
const closes = (tpl.match(/\}/g) ?? []).length;
expect(opens).toBe(closes);
});
it("ends with export default workflow", () => {
const tpl = buildWorkflowTemplate("test");
expect(tpl.trim().endsWith("export default workflow;")).toBe(true);
});
});
describe("workflow scaffold file writing (simulated)", () => {
it("writes the template to disk correctly", () => {
const { mkdirSync, writeFileSync } = require("node:fs");
const workflowDir = join(tmpDir, "workflows", "my-task");
mkdirSync(workflowDir, { recursive: true });
const content = buildWorkflowTemplate("my-task");
writeFileSync(join(workflowDir, "index.ts"), content, "utf8");
const read = readFileSync(join(workflowDir, "index.ts"), "utf8");
expect(read).toContain("my-task started");
expect(read).toContain("WorkflowDefinition");
});
});
+250
View File
@@ -0,0 +1,250 @@
/**
* Tests for nerve logs command — pure helper functions only.
*
* We test sliceLogs and buildLogFooter without touching the filesystem or
* spawning a real process.
*/
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { DEFAULT_LOG_LINES, buildLogFooter, readAllLines, sliceLogs } from "../commands/logs.js";
import { logsCommand } from "../commands/logs.js";
// ---------------------------------------------------------------------------
// sliceLogs
// ---------------------------------------------------------------------------
describe("sliceLogs", () => {
const make = (n: number) => Array.from({ length: n }, (_, i) => `line ${i + 1}`);
it("returns empty result for empty array", () => {
const r = sliceLogs([], 0, 50);
expect(r.lines).toHaveLength(0);
expect(r.total).toBe(0);
expect(r.nextOffset).toBeNull();
});
it("tail mode (offset=0): returns last N lines", () => {
const lines = make(100);
const r = sliceLogs(lines, 0, 10);
expect(r.lines).toHaveLength(10);
expect(r.lines[0]).toBe("line 91");
expect(r.lines[9]).toBe("line 100");
expect(r.startLine).toBe(91);
expect(r.endLine).toBe(100);
});
it("tail mode: when file shorter than limit, returns all", () => {
const lines = make(20);
const r = sliceLogs(lines, 0, 50);
expect(r.lines).toHaveLength(20);
expect(r.startLine).toBe(1);
expect(r.endLine).toBe(20);
expect(r.nextOffset).toBeNull();
});
it("tail mode: provides nextOffset when earlier lines exist", () => {
const lines = make(200);
const r = sliceLogs(lines, 0, 50);
expect(r.nextOffset).not.toBeNull();
expect(r.nextOffset).toBe(151 - 50); // startLine=151, prev page starts at 101
});
it("tail mode: nextOffset is null when showing from line 1", () => {
const lines = make(40);
const r = sliceLogs(lines, 0, 50);
expect(r.nextOffset).toBeNull();
});
it("offset mode: starts at given 1-based line number", () => {
const lines = make(100);
const r = sliceLogs(lines, 10, 5);
expect(r.lines[0]).toBe("line 10");
expect(r.startLine).toBe(10);
expect(r.endLine).toBe(14);
});
it("offset mode: clamps start to 0 for offset=1", () => {
const lines = make(50);
const r = sliceLogs(lines, 1, 10);
expect(r.startLine).toBe(1);
});
it("offset mode: nextOffset is null when slice starts at line 1", () => {
const lines = make(50);
const r = sliceLogs(lines, 1, 20);
expect(r.nextOffset).toBeNull();
});
it("offset mode: nextOffset points to previous page", () => {
const lines = make(100);
const r = sliceLogs(lines, 51, 50); // lines 51-100
expect(r.nextOffset).toBe(1); // previous page starts at line 1
});
});
// ---------------------------------------------------------------------------
// buildLogFooter
// ---------------------------------------------------------------------------
describe("buildLogFooter", () => {
it("returns empty-file message when total=0", () => {
const slice = { lines: [], total: 0, startLine: 0, endLine: 0, nextOffset: null };
expect(buildLogFooter(slice, 50, "/path/to/nerve.log")).toContain("empty");
});
it("includes range and path in footer", () => {
const slice = { lines: ["x"], total: 200, startLine: 151, endLine: 200, nextOffset: 101 };
const footer = buildLogFooter(slice, 50, "/var/log/nerve.log");
expect(footer).toContain("lines 151-200 of 200");
expect(footer).toContain("/var/log/nerve.log");
});
it("includes pagination hint when nextOffset is set", () => {
const slice = { lines: ["x"], total: 200, startLine: 151, endLine: 200, nextOffset: 101 };
const footer = buildLogFooter(slice, 50, "/path/nerve.log");
expect(footer).toContain("nerve logs --offset 101 -n 50");
});
it("no pagination hint when nextOffset is null", () => {
const slice = { lines: ["x"], total: 20, startLine: 1, endLine: 20, nextOffset: null };
const footer = buildLogFooter(slice, 50, "/path/nerve.log");
expect(footer).not.toContain("nerve logs --offset");
});
});
// ---------------------------------------------------------------------------
// DEFAULT_LOG_LINES constant
// ---------------------------------------------------------------------------
describe("DEFAULT_LOG_LINES", () => {
it("is 50", () => {
expect(DEFAULT_LOG_LINES).toBe(50);
});
});
// ---------------------------------------------------------------------------
// readAllLines
// ---------------------------------------------------------------------------
describe("readAllLines", () => {
let tmpDir: string;
beforeEach(() => {
tmpDir = mkdtempSync(join(tmpdir(), "nerve-logs-test-"));
});
afterEach(() => {
rmSync(tmpDir, { recursive: true, force: true });
});
it("returns empty array for nonexistent file", async () => {
const result = await readAllLines(join(tmpDir, "missing.log"));
expect(result).toHaveLength(0);
});
it("reads all lines from a file", async () => {
const logFile = join(tmpDir, "test.log");
writeFileSync(logFile, "line1\nline2\nline3\n");
const result = await readAllLines(logFile);
expect(result).toEqual(["line1", "line2", "line3"]);
});
it("handles file with no trailing newline", async () => {
const logFile = join(tmpDir, "test.log");
writeFileSync(logFile, "a\nb\nc");
const result = await readAllLines(logFile);
expect(result).toEqual(["a", "b", "c"]);
});
it("returns empty array for empty file", async () => {
const logFile = join(tmpDir, "empty.log");
writeFileSync(logFile, "");
const result = await readAllLines(logFile);
expect(result).toHaveLength(0);
});
});
// ---------------------------------------------------------------------------
// Integration: readAllLines + sliceLogs end-to-end
// ---------------------------------------------------------------------------
describe("readAllLines + sliceLogs integration", () => {
let tmpDir: string;
beforeEach(() => {
tmpDir = mkdtempSync(join(tmpdir(), "nerve-logs-int-"));
});
afterEach(() => {
rmSync(tmpDir, { recursive: true, force: true });
});
it("tail-paginates a large log file correctly", async () => {
const logFile = join(tmpDir, "big.log");
const content = Array.from({ length: 120 }, (_, i) => `entry ${i + 1}`).join("\n");
writeFileSync(logFile, content);
const all = await readAllLines(logFile);
const page1 = sliceLogs(all, 0, 50); // last 50: lines 71-120
expect(page1.startLine).toBe(71);
expect(page1.endLine).toBe(120);
expect(page1.nextOffset).toBe(21); // max(1, 71-50)
const page2 = sliceLogs(all, page1.nextOffset!, 50); // lines 21-70
expect(page2.startLine).toBe(21);
expect(page2.endLine).toBe(70);
expect(page2.nextOffset).toBe(1); // max(1, 21-50) = 1
const page3 = sliceLogs(all, page2.nextOffset!, 50); // lines 1-50
expect(page3.startLine).toBe(1);
expect(page3.endLine).toBe(50);
expect(page3.nextOffset).toBeNull();
});
});
// ---------------------------------------------------------------------------
// logsCommand: negative offset validation
// ---------------------------------------------------------------------------
describe("logsCommand negative offset", () => {
let stderrOutput: string;
let exitCode: number | undefined;
beforeEach(() => {
stderrOutput = "";
exitCode = undefined;
vi.spyOn(process.stderr, "write").mockImplementation((chunk) => {
stderrOutput += typeof chunk === "string" ? chunk : chunk.toString();
return true;
});
vi.spyOn(process, "exit").mockImplementation((code?: number | string | null) => {
exitCode = typeof code === "number" ? code : 1;
throw new Error(`process.exit(${exitCode})`);
});
});
afterEach(() => {
vi.restoreAllMocks();
});
it("exits with code 1 and writes to stderr when offset is negative", async () => {
await expect(
logsCommand.run!({ args: { n: "50", offset: "-5", follow: false }, rawArgs: [], cmd: logsCommand as never }),
).rejects.toThrow("process.exit(1)");
expect(exitCode).toBe(1);
expect(stderrOutput).toContain("--offset must be a non-negative integer");
expect(stderrOutput).toContain("-5");
});
it("exits with code 1 for offset=-1", async () => {
await expect(
logsCommand.run!({ args: { n: "10", offset: "-1", follow: false }, rawArgs: [], cmd: logsCommand as never }),
).rejects.toThrow("process.exit(1)");
expect(exitCode).toBe(1);
});
});
@@ -0,0 +1,301 @@
/**
* Tests for `nerve sense list` — formatting helpers and IPC round-trip.
*
* Covers:
* - formatDuration helper
* - formatSenseList output
* - sensesFromConfig (static fallback from nerve.yaml)
* - listSensesViaDaemon IPC round-trip via real Unix socket
*/
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
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 { 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: "disk-usage", group: "system", throttle: 30000, timeout: null, lastSignalTs: null },
{ name: "active-tasks", group: "tasks", throttle: 10000, timeout: 30000, lastSignalTs: null },
];
// ---------------------------------------------------------------------------
// formatDuration
// ---------------------------------------------------------------------------
describe("formatDuration", () => {
it("returns '—' for null", () => {
expect(formatDuration(null)).toBe("—");
});
it("formats sub-minute durations as seconds", () => {
expect(formatDuration(0)).toBe("0s");
expect(formatDuration(1000)).toBe("1s");
expect(formatDuration(59000)).toBe("59s");
});
it("formats minute-range durations as Xm Ys", () => {
expect(formatDuration(60000)).toBe("1m 0s");
expect(formatDuration(90000)).toBe("1m 30s");
expect(formatDuration(3599000)).toBe("59m 59s");
});
it("formats hour-range durations as Xh Ym", () => {
expect(formatDuration(3600000)).toBe("1h 0m");
expect(formatDuration(3660000)).toBe("1h 1m");
expect(formatDuration(7200000)).toBe("2h 0m");
});
});
// ---------------------------------------------------------------------------
// formatSenseList
// ---------------------------------------------------------------------------
describe("formatSenseList", () => {
it("returns empty message when no senses", () => {
const output = formatSenseList([]);
expect(output).toContain("No senses registered");
});
it("shows sense count in header", () => {
const output = formatSenseList(SAMPLE_SENSES);
expect(output).toContain("3");
});
it("shows each sense name", () => {
const output = formatSenseList(SAMPLE_SENSES);
expect(output).toContain("cpu-usage");
expect(output).toContain("disk-usage");
expect(output).toContain("active-tasks");
});
it("shows group for each sense", () => {
const output = formatSenseList(SAMPLE_SENSES);
expect(output).toContain("system");
expect(output).toContain("tasks");
});
it("shows throttle and timeout durations", () => {
const output = formatSenseList(SAMPLE_SENSES);
// cpu-usage: throttle=5s, timeout=3s
expect(output).toContain("5s");
expect(output).toContain("3s");
// disk-usage: timeout=null → '—'
expect(output).toContain("—");
});
it("shows '(never)' when lastSignalTs is null", () => {
const output = formatSenseList(SAMPLE_SENSES);
expect(output).toContain("(never)");
});
it("shows ISO timestamp when lastSignalTs is set", () => {
const output = formatSenseList(SAMPLE_SENSES);
// cpu-usage has lastSignalTs = 1_700_000_000_000
expect(output).toContain(new Date(1_700_000_000_000).toISOString());
});
});
// ---------------------------------------------------------------------------
// sensesFromConfig — static fallback from nerve.yaml
// ---------------------------------------------------------------------------
describe("sensesFromConfig", () => {
let tmpDir: string;
beforeEach(() => {
tmpDir = mkdtempSync(join(tmpdir(), "nerve-sense-list-test-"));
});
afterEach(() => {
rmSync(tmpDir, { recursive: true, force: true });
});
it("returns empty array when file does not exist", () => {
const result = sensesFromConfig(join(tmpDir, "nonexistent.yaml"));
expect(result).toEqual([]);
});
it("returns empty array when file has invalid YAML", () => {
const path = join(tmpDir, "nerve.yaml");
writeFileSync(path, "not: valid: yaml: :::");
const result = sensesFromConfig(path);
expect(result).toEqual([]);
});
it("parses senses from valid nerve.yaml", () => {
const path = join(tmpDir, "nerve.yaml");
writeFileSync(
path,
`
senses:
cpu-usage:
group: system
throttle: 5s
timeout: 3s
disk-usage:
group: system
throttle: 30s
reflexes: []
`.trim(),
);
const result = sensesFromConfig(path);
expect(result).toHaveLength(2);
expect(result[0]).toMatchObject({ name: "cpu-usage", group: "system", lastSignalTs: null });
expect(result[1]).toMatchObject({ name: "disk-usage", group: "system", lastSignalTs: null });
});
it("always sets lastSignalTs to null (static fallback)", () => {
const path = join(tmpDir, "nerve.yaml");
writeFileSync(
path,
`
senses:
my-sense:
group: default
reflexes: []
`.trim(),
);
const result = sensesFromConfig(path);
expect(result[0].lastSignalTs).toBeNull();
});
it("populates throttle and timeout from config", () => {
const path = join(tmpDir, "nerve.yaml");
writeFileSync(
path,
`
senses:
my-sense:
group: default
throttle: 10s
timeout: 5s
reflexes: []
`.trim(),
);
const result = sensesFromConfig(path);
expect(result[0].throttle).toBe(10000);
expect(result[0].timeout).toBe(5000);
});
});
// ---------------------------------------------------------------------------
// listSensesViaDaemon — IPC round-trip via real Unix socket
// ---------------------------------------------------------------------------
describe("listSensesViaDaemon", () => {
let sockDir: string;
let sockPath: string;
beforeEach(() => {
sockDir = mkdtempSync(join(tmpdir(), "nerve-sense-list-ipc-"));
sockPath = join(sockDir, "nerve.sock");
});
afterEach(() => {
rmSync(sockDir, { recursive: true, force: true });
});
it("resolves with { ok: true, senses: [] } when daemon returns empty list", async () => {
const server = createServer((s) => {
s.on("data", (chunk: Buffer) => {
const line = chunk.toString("utf8").trim();
try {
const req = JSON.parse(line) as { type: string };
if (req.type === "list-senses") {
s.write(`${JSON.stringify({ ok: true, senses: [] })}\n`);
}
} catch {
// ignore
}
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await listSensesViaDaemon(sockPath);
expect(result).toEqual({ ok: true, senses: [] });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
it("resolves with populated senses array", async () => {
const senses: SenseInfo[] = [
{ name: "cpu-usage", group: "system", throttle: 5000, timeout: 3000, lastSignalTs: 12345 },
];
const server = createServer((s) => {
s.on("data", () => {
s.write(`${JSON.stringify({ ok: true, senses })}\n`);
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await listSensesViaDaemon(sockPath);
expect(result).toEqual({ ok: true, senses });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
it("resolves with { ok: false, error } when daemon returns an error", async () => {
const server = createServer((s) => {
s.on("data", () => {
s.write(`${JSON.stringify({ ok: false, error: "something went wrong" })}\n`);
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await listSensesViaDaemon(sockPath);
expect(result).toEqual({ ok: false, error: "something went wrong" });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
it("rejects when no daemon is listening on the socket", async () => {
await expect(listSensesViaDaemon(sockPath)).rejects.toThrow(/Cannot connect to daemon/);
});
it("sends a list-senses IPC message to the daemon", async () => {
const received: unknown[] = [];
const server = createServer((s) => {
s.on("data", (chunk: Buffer) => {
const line = chunk.toString("utf8").trim();
try {
received.push(JSON.parse(line));
} catch {
// ignore
}
s.write(`${JSON.stringify({ ok: true, senses: [] })}\n`);
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
await listSensesViaDaemon(sockPath);
expect(received).toHaveLength(1);
expect(received[0]).toMatchObject({ type: "list-senses" });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
});
+110
View File
@@ -0,0 +1,110 @@
/**
* Tests for the sense CLI helper — triggerSenseViaDaemon IPC round-trip.
*
* Uses a real Unix socket server to validate the full client/server
* protocol without requiring a running daemon process.
*/
import { mkdtempSync, rmSync } from "node:fs";
import { createServer } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { triggerSenseViaDaemon } from "../daemon-client.js";
// ---------------------------------------------------------------------------
// Test setup
// ---------------------------------------------------------------------------
let sockDir: string;
let sockPath: string;
beforeEach(() => {
sockDir = mkdtempSync(join(tmpdir(), "nerve-sense-test-"));
sockPath = join(sockDir, "nerve.sock");
});
afterEach(() => {
rmSync(sockDir, { recursive: true, force: true });
});
// ---------------------------------------------------------------------------
// triggerSenseViaDaemon — IPC round-trip via real Unix socket
// ---------------------------------------------------------------------------
describe("triggerSenseViaDaemon", () => {
it("resolves { ok: true } when daemon responds ok", async () => {
const received: unknown[] = [];
const server = createServer((s) => {
s.on("data", (chunk: Buffer) => {
const line = chunk.toString("utf8").trim();
try {
received.push(JSON.parse(line));
} catch {
// ignore
}
s.write(`${JSON.stringify({ ok: true })}\n`);
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await triggerSenseViaDaemon(sockPath, "cpu-usage");
expect(result).toEqual({ ok: true });
// Verify the correct IPC message was sent
expect(received).toHaveLength(1);
expect(received[0]).toMatchObject({ type: "trigger-sense", sense: "cpu-usage" });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
it("resolves { ok: false, error } when daemon rejects the sense", async () => {
const server = createServer((s) => {
s.on("data", () => {
s.write(`${JSON.stringify({ ok: false, error: 'Unknown sense: "no-such-sense"' })}\n`);
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await triggerSenseViaDaemon(sockPath, "no-such-sense");
expect(result).toEqual({ ok: false, error: 'Unknown sense: "no-such-sense"' });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
it("rejects when no daemon is listening on the socket", async () => {
await expect(triggerSenseViaDaemon(sockPath, "cpu-usage")).rejects.toThrow(
/Cannot connect to daemon/,
);
});
it("sends the sense name exactly as provided", async () => {
const received: unknown[] = [];
const server = createServer((s) => {
s.on("data", (chunk: Buffer) => {
const line = chunk.toString("utf8").trim();
try {
received.push(JSON.parse(line));
} catch {
// ignore
}
s.write(`${JSON.stringify({ ok: true })}\n`);
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
await triggerSenseViaDaemon(sockPath, "my-custom-sense");
expect(received[0]).toMatchObject({ sense: "my-custom-sense" });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
});
+449
View File
@@ -0,0 +1,449 @@
/**
* Tests for workflow CLI commands — pure logic helpers.
*
* Tests do NOT invoke the citty command handlers directly (they would call
* process.exit / process.stdout.write against a real terminal). Instead we
* test the exported pure helper functions that the command handlers delegate
* to. The helpers use real LogStore / SQLite via temp directories.
*/
import { mkdtempSync, rmSync } from "node:fs";
import { createServer } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createLogStore } from "@uncaged/nerve-daemon";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import {
buildInspectOutput,
buildListOutput,
formatTs,
getAllWorkflowRuns,
parseIntArg,
statusIcon,
} from "../commands/workflow.js";
import { triggerWorkflowViaDaemon } from "../daemon-client.js";
import type { LogStore, WorkflowRun } from "../daemon-types.js";
// ---------------------------------------------------------------------------
// Test helpers
// ---------------------------------------------------------------------------
let tmpDir: string;
let store: LogStore;
function upsertRun(
runId: string,
workflow: string,
status: WorkflowRun["status"],
ts: number,
): void {
store.upsertWorkflowRun(
{ source: "workflow", type: status, refId: runId, payload: null, ts },
{ runId, workflow, status, ts },
);
}
beforeEach(() => {
tmpDir = mkdtempSync(join(tmpdir(), "nerve-cli-wf-test-"));
store = createLogStore(join(tmpDir, "data", "logs.db"));
});
afterEach(() => {
store.close();
rmSync(tmpDir, { recursive: true, force: true });
});
// ---------------------------------------------------------------------------
// formatTs
// ---------------------------------------------------------------------------
describe("formatTs", () => {
it("returns ISO 8601 string", () => {
const ts = new Date("2026-01-01T00:00:00.000Z").getTime();
expect(formatTs(ts)).toBe("2026-01-01T00:00:00.000Z");
});
});
// ---------------------------------------------------------------------------
// statusIcon
// ---------------------------------------------------------------------------
describe("statusIcon", () => {
it.each([
["started", "▶"],
["queued", "⏳"],
["completed", "✅"],
["failed", "❌"],
["crashed", "💥"],
["dropped", "🗑"],
["interrupted", "⚠️"],
] as const)("maps status=%s to icon=%s", (status, icon) => {
expect(statusIcon(status)).toBe(icon);
});
});
// ---------------------------------------------------------------------------
// getAllWorkflowRuns
// ---------------------------------------------------------------------------
describe("getAllWorkflowRuns", () => {
it("returns empty array when no runs exist", () => {
expect(getAllWorkflowRuns(store, null)).toHaveLength(0);
});
it("returns all runs across statuses", () => {
upsertRun("r1", "cleanup", "completed", 1000);
upsertRun("r2", "cleanup", "started", 2000);
upsertRun("r3", "deploy", "failed", 1500);
const runs = getAllWorkflowRuns(store, null);
expect(runs).toHaveLength(3);
});
it("deduplicates runs by runId (latest state only)", () => {
upsertRun("r1", "cleanup", "started", 1000);
upsertRun("r1", "cleanup", "completed", 2000);
const runs = getAllWorkflowRuns(store, null);
expect(runs).toHaveLength(1);
expect(runs[0].status).toBe("completed");
});
it("filters by workflow name", () => {
upsertRun("r1", "cleanup", "completed", 1000);
upsertRun("r2", "deploy", "started", 2000);
upsertRun("r3", "cleanup", "failed", 1500);
const runs = getAllWorkflowRuns(store, "cleanup");
expect(runs).toHaveLength(2);
for (const r of runs) {
expect(r.workflow).toBe("cleanup");
}
});
it("sorts by ts descending (newest first)", () => {
upsertRun("r1", "cleanup", "completed", 1000);
upsertRun("r2", "cleanup", "started", 3000);
upsertRun("r3", "cleanup", "failed", 2000);
const runs = getAllWorkflowRuns(store, null);
expect(runs[0].ts).toBeGreaterThan(runs[1].ts);
expect(runs[1].ts).toBeGreaterThan(runs[2].ts);
});
});
// ---------------------------------------------------------------------------
// buildListOutput
// ---------------------------------------------------------------------------
describe("buildListOutput", () => {
function makeRun(
runId: string,
workflow: string,
status: WorkflowRun["status"],
ts: number,
): WorkflowRun {
return { runId, workflow, status, ts };
}
it("returns empty message when no runs and --all=false", () => {
const { lines, paginationHint } = buildListOutput([], 0, 20, false, null);
expect(lines).toHaveLength(1);
expect(lines[0]).toContain("--all");
expect(paginationHint).toBeNull();
});
it("returns empty message when no runs and --all=true", () => {
const { lines, paginationHint } = buildListOutput([], 0, 20, true, null);
expect(lines).toHaveLength(1);
expect(lines[0]).not.toContain("--all");
expect(paginationHint).toBeNull();
});
it("shows correct run count in header", () => {
const runs = [
makeRun("r1", "cleanup", "started", 1000),
makeRun("r2", "cleanup", "queued", 2000),
];
const { lines } = buildListOutput(runs, 0, 20, false, null);
expect(lines[0]).toContain("2 of 2");
});
it("includes run details in lines", () => {
const runs = [makeRun("run-abc", "my-workflow", "started", 1000)];
const { lines } = buildListOutput(runs, 0, 20, false, null);
const combined = lines.join("");
expect(combined).toContain("run-abc");
expect(combined).toContain("my-workflow");
expect(combined).toContain("started");
expect(combined).toContain("▶");
});
it("paginates: shows only limit entries and provides hint", () => {
const runs = Array.from({ length: 5 }, (_, i) => makeRun(`r${i}`, "wf", "completed", i * 1000));
const { lines, paginationHint } = buildListOutput(runs, 0, 2, true, null);
// header + 2 run lines
expect(lines).toHaveLength(3);
expect(paginationHint).not.toBeNull();
expect(paginationHint).toContain("--offset 2");
expect(paginationHint).toContain("3 more");
});
it("pagination hint includes --all flag when set", () => {
const runs = Array.from({ length: 3 }, (_, i) => makeRun(`r${i}`, "wf", "completed", i * 1000));
const { paginationHint } = buildListOutput(runs, 0, 1, true, null);
expect(paginationHint).toContain("--all");
});
it("pagination hint includes --workflow filter when set", () => {
const runs = Array.from({ length: 3 }, (_, i) =>
makeRun(`r${i}`, "cleanup", "completed", i * 1000),
);
const { paginationHint } = buildListOutput(runs, 0, 1, false, "cleanup");
expect(paginationHint).toContain("--workflow cleanup");
});
it("no pagination hint when all entries fit on one page", () => {
const runs = [makeRun("r1", "wf", "started", 1000)];
const { paginationHint } = buildListOutput(runs, 0, 20, false, null);
expect(paginationHint).toBeNull();
});
it("respects offset for pagination", () => {
const runs = Array.from({ length: 5 }, (_, i) => makeRun(`r${i}`, "wf", "completed", i * 1000));
const { lines, paginationHint } = buildListOutput(runs, 2, 2, true, null);
// header + 2 run lines (offset=2, limit=2 gives items 2 and 3)
expect(lines).toHaveLength(3);
// 1 item remaining (index 4)
expect(paginationHint).toContain("1 more");
expect(paginationHint).toContain("--offset 4");
});
});
// ---------------------------------------------------------------------------
// buildInspectOutput
// ---------------------------------------------------------------------------
describe("buildInspectOutput", () => {
const baseRun: WorkflowRun = {
runId: "run-xyz",
workflow: "cleanup",
status: "completed",
ts: 1_700_000_000_000,
};
it("shows header with run details", () => {
const { header } = buildInspectOutput(baseRun, [], 0, 20);
const headerText = header.join("");
expect(headerText).toContain("run-xyz");
expect(headerText).toContain("cleanup");
expect(headerText).toContain("completed");
});
it("shows '(no events recorded)' when log is empty", () => {
const { eventLines } = buildInspectOutput(baseRun, [], 0, 20);
expect(eventLines.join("")).toContain("no events recorded");
});
it("shows event lines with type and ts", () => {
const logs = [{ ts: 1_700_000_001_000, type: "started", payload: null }];
const { eventLines } = buildInspectOutput(baseRun, logs, 0, 20);
const text = eventLines.join("");
expect(text).toContain("type=started");
});
it("truncates long payloads to 200 chars with ellipsis", () => {
const longPayload = "x".repeat(250);
const logs = [{ ts: 1000, type: "step_complete", payload: longPayload }];
const { eventLines } = buildInspectOutput(baseRun, logs, 0, 20);
const text = eventLines.join("");
expect(text).toContain("…");
expect(text).not.toContain("x".repeat(201));
});
it("shows short payloads in full", () => {
const logs = [{ ts: 1000, type: "step_complete", payload: '{"count":5}' }];
const { eventLines } = buildInspectOutput(baseRun, logs, 0, 20);
expect(eventLines.join("")).toContain('{"count":5}');
});
it("paginates events with a hint", () => {
const logs = Array.from({ length: 5 }, (_, i) => ({
ts: 1000 + i,
type: "step_complete",
payload: null,
}));
const { eventLines, paginationHint } = buildInspectOutput(baseRun, logs, 0, 2);
expect(eventLines).toHaveLength(2);
expect(paginationHint).toContain("3 more");
expect(paginationHint).toContain("--offset 2");
expect(paginationHint).toContain("run-xyz");
});
it("no pagination hint when all events fit on one page", () => {
const logs = [{ ts: 1000, type: "started", payload: null }];
const { paginationHint } = buildInspectOutput(baseRun, logs, 0, 20);
expect(paginationHint).toBeNull();
});
});
// ---------------------------------------------------------------------------
// Integration: getAllWorkflowRuns + buildListOutput end-to-end with real store
// ---------------------------------------------------------------------------
describe("workflow list — integration with real store", () => {
it("lists active runs from the store", () => {
upsertRun("r1", "cleanup", "started", 1000);
upsertRun("r2", "cleanup", "queued", 2000);
upsertRun("r3", "cleanup", "completed", 3000);
// Active only (getActiveWorkflowRuns)
const activeRuns = store.getActiveWorkflowRuns();
const { lines } = buildListOutput(activeRuns, 0, 20, false, null);
const combined = lines.join("");
expect(combined).toContain("r1");
expect(combined).toContain("r2");
expect(combined).not.toContain("r3");
});
it("lists all runs with getAllWorkflowRuns", () => {
upsertRun("r1", "cleanup", "started", 1000);
upsertRun("r2", "cleanup", "completed", 2000);
upsertRun("r3", "cleanup", "failed", 3000);
const allRuns = getAllWorkflowRuns(store, null);
const { lines } = buildListOutput(allRuns, 0, 20, true, null);
const combined = lines.join("");
expect(combined).toContain("r1");
expect(combined).toContain("r2");
expect(combined).toContain("r3");
});
});
// ---------------------------------------------------------------------------
// parseIntArg
// ---------------------------------------------------------------------------
describe("parseIntArg", () => {
it("parses a valid integer string", () => {
expect(parseIntArg("5", 20)).toBe(5);
});
it("returns fallback for non-numeric string", () => {
expect(parseIntArg("abc", 20)).toBe(20);
});
it("returns the value for '0' (not fallback)", () => {
expect(parseIntArg("0", 20)).toBe(0);
});
it("returns fallback for empty string", () => {
expect(parseIntArg("", 20)).toBe(20);
});
it("parses negative integers", () => {
expect(parseIntArg("-3", 20)).toBe(-3);
});
});
// ---------------------------------------------------------------------------
// getAllWorkflowRuns — backed by real store's SQL query
// ---------------------------------------------------------------------------
describe("getAllWorkflowRuns — uses store.getAllWorkflowRuns SQL path", () => {
it("returns all runs regardless of status", () => {
upsertRun("r1", "deploy", "completed", 1000);
upsertRun("r2", "deploy", "failed", 2000);
upsertRun("r3", "deploy", "started", 3000);
upsertRun("r4", "deploy", "queued", 4000);
upsertRun("r5", "deploy", "crashed", 5000);
upsertRun("r6", "deploy", "dropped", 6000);
upsertRun("r7", "deploy", "interrupted", 7000);
const runs = getAllWorkflowRuns(store, null);
expect(runs).toHaveLength(7);
});
it("returns runs sorted by ts descending (newest first)", () => {
upsertRun("r1", "deploy", "completed", 1000);
upsertRun("r2", "deploy", "completed", 3000);
upsertRun("r3", "deploy", "completed", 2000);
const runs = getAllWorkflowRuns(store, null);
expect(runs[0].ts).toBe(3000);
expect(runs[1].ts).toBe(2000);
expect(runs[2].ts).toBe(1000);
});
it("filters by workflow name", () => {
upsertRun("r1", "alpha", "completed", 1000);
upsertRun("r2", "beta", "completed", 2000);
upsertRun("r3", "alpha", "failed", 3000);
const runs = getAllWorkflowRuns(store, "alpha");
expect(runs).toHaveLength(2);
for (const r of runs) expect(r.workflow).toBe("alpha");
});
it("returns empty array when store has no runs", () => {
expect(getAllWorkflowRuns(store, null)).toHaveLength(0);
});
});
// ---------------------------------------------------------------------------
// triggerWorkflowViaDaemon — IPC round-trip via real Unix socket
// ---------------------------------------------------------------------------
describe("triggerWorkflowViaDaemon", () => {
let sockDir: string;
let sockPath: string;
beforeEach(() => {
sockDir = mkdtempSync(join(tmpdir(), "nerve-ipc-test-"));
sockPath = join(sockDir, "nerve.sock");
});
afterEach(() => {
rmSync(sockDir, { recursive: true, force: true });
});
it("resolves { ok: true } when server responds ok", async () => {
const server = createServer((s) => {
s.on("data", () => {
s.write(`${JSON.stringify({ ok: true })}\n`);
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await triggerWorkflowViaDaemon(sockPath, "my-workflow", {});
expect(result).toEqual({ ok: true });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
it("resolves { ok: false, error } when server responds with error", async () => {
const server = createServer((s) => {
s.on("data", () => {
s.write(`${JSON.stringify({ ok: false, error: "unknown workflow" })}\n`);
});
});
await new Promise<void>((r) => server.listen(sockPath, r));
try {
const result = await triggerWorkflowViaDaemon(sockPath, "missing", {});
expect(result).toEqual({ ok: false, error: "unknown workflow" });
} finally {
await new Promise<void>((r) => server.close(() => r()));
}
});
it("rejects when no daemon is listening on the socket", async () => {
await expect(triggerWorkflowViaDaemon(sockPath, "my-workflow", {})).rejects.toThrow(
/Cannot connect to daemon/,
);
});
});
+14 -4
View File
@@ -1,12 +1,16 @@
#!/usr/bin/env node
import { defineCommand, runMain } from "citty";
import { daemonCommand } from "./commands/daemon.js";
import { devCommand } from "./commands/dev.js";
import { initCommand } from "./commands/init.js";
import { startCommand } from "./commands/start.js";
import { logsCommand } from "./commands/logs.js";
import { senseCommand } from "./commands/sense.js";
import { daemonStartCommand } from "./commands/start.js";
import { statusCommand } from "./commands/status.js";
import { stopCommand } from "./commands/stop.js";
import { storeCommand } from "./commands/store.js";
import { validateCommand } from "./commands/validate.js";
import { workflowCommand } from "./commands/workflow.js";
const main = defineCommand({
meta: {
@@ -15,10 +19,16 @@ const main = defineCommand({
},
subCommands: {
init: initCommand,
start: startCommand,
daemon: daemonCommand,
dev: devCommand,
start: daemonStartCommand,
stop: stopCommand,
status: statusCommand,
logs: logsCommand,
validate: validateCommand,
sense: senseCommand,
store: storeCommand,
workflow: workflowCommand,
},
});
+31
View File
@@ -0,0 +1,31 @@
import { defineCommand } from "citty";
import { logsCommand } from "./logs.js";
import { daemonStartCommand, runDaemonStartCommand } from "./start.js";
import { statusCommand } from "./status.js";
import { runStopCommand, stopCommand } from "./stop.js";
const daemonRestartCommand = defineCommand({
meta: {
name: "restart",
description: "Stop then start the nerve daemon",
},
async run() {
await runStopCommand();
await runDaemonStartCommand();
},
});
export const daemonCommand = defineCommand({
meta: {
name: "daemon",
description: "Manage the nerve background daemon",
},
subCommands: {
start: daemonStartCommand,
stop: stopCommand,
status: statusCommand,
restart: daemonRestartCommand,
logs: logsCommand,
},
});
+17
View File
@@ -0,0 +1,17 @@
import { defineCommand } from "citty";
import { runForegroundKernelSession } from "../run-foreground-kernel.js";
import { loadDaemonModule } from "../workspace-daemon.js";
import { getNerveRoot } from "../workspace.js";
export const devCommand = defineCommand({
meta: {
name: "dev",
description: "Run the nerve kernel in the foreground (development mode)",
},
async run() {
const nerveRoot = getNerveRoot();
const { createKernel } = await loadDaemonModule(nerveRoot);
await runForegroundKernelSession(nerveRoot, createKernel);
},
});
+207 -48
View File
@@ -26,10 +26,14 @@ const PACKAGE_JSON = `{
"type": "module",
"dependencies": {
"@uncaged/nerve-core": "latest",
"@uncaged/nerve-daemon": "latest",
"drizzle-orm": "latest"
},
"devDependencies": {
"drizzle-kit": "latest"
},
"pnpm": {
"onlyBuiltDependencies": ["better-sqlite3", "esbuild"]
}
}
`;
@@ -48,9 +52,9 @@ export const cpuUsage = sqliteTable("cpu_usage", {
});
`;
const CPU_INDEX_TS = `import { cpus } from "node:os";
const CPU_INDEX_JS = `import { cpus } from "node:os";
export async function compute(): Promise<unknown> {
export async function compute() {
const cpuList = cpus();
let totalIdle = 0;
@@ -97,7 +101,7 @@ async function runCommand(cmd: string, args: string[], cwd: string): Promise<voi
});
}
async function detectPackageManager(): Promise<{ cmd: string; args: string[] }> {
async function detectPackageManager(): Promise<{ cmd: string; installArgs: string[] }> {
const { execFile } = await import("node:child_process");
const { promisify } = await import("node:util");
const execFileAsync = promisify(execFile);
@@ -105,19 +109,103 @@ async function detectPackageManager(): Promise<{ cmd: string; args: string[] }>
for (const pm of ["pnpm", "yarn", "npm"]) {
try {
await execFileAsync(pm, ["--version"]);
const args = pm === "pnpm" ? ["install", "--no-cache"] : ["install"];
return { cmd: pm, args };
const installArgs = pm === "pnpm" ? ["install", "--no-cache"] : ["install"];
return { cmd: pm, installArgs };
} catch {
// not available, try next
}
}
return { cmd: "npm", args: ["install"] };
return { cmd: "npm", installArgs: ["install"] };
}
export const initCommand = defineCommand({
export const WORKFLOW_NAME_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$|^[a-z0-9]$/;
export function validateWorkflowName(name: string): string | null {
if (name.length === 0) return "Workflow name must not be empty.";
if (name.length > 64) return "Workflow name must be 64 characters or fewer.";
if (!WORKFLOW_NAME_RE.test(name))
return "Workflow name must contain only lowercase letters, digits, and hyphens, and must not start or end with a hyphen.";
return null;
}
export function buildWorkflowTemplate(name: string): string {
return `import type { WorkflowDefinition } from "@uncaged/nerve-daemon";
const workflow: WorkflowDefinition = {
roles: {
main: {
async execute(prompt, ctx) {
ctx.log("${name} started");
// TODO: implement your role logic here
return { type: "done" };
},
},
},
moderate(thread, event) {
if (event.type === "thread_start") {
return { role: "main", prompt: {} };
}
return null; // workflow complete
},
};
export default workflow;
`;
}
const initWorkflowCommand = defineCommand({
meta: {
name: "init",
description: "Initialize the ~/.uncaged-nerve/ workspace",
name: "workflow",
description: "Scaffold a new workflow template in ~/.uncaged-nerve/workflows/<name>/",
},
args: {
name: {
type: "positional",
description: "Workflow name (must match the key in nerve.yaml workflows section)",
},
force: {
type: "boolean",
description: "Overwrite if the workflow directory already exists",
default: false,
},
},
async run({ args }) {
const nerveRoot = getNerveRoot();
const workflowDir = join(nerveRoot, "workflows", args.name);
const nameError = validateWorkflowName(args.name);
if (nameError !== null) {
process.stderr.write(`❌ Invalid workflow name: ${nameError}\n`);
process.exit(1);
}
if (existsSync(workflowDir) && !args.force) {
process.stderr.write(
`⚠️ Workflow "${args.name}" already exists at ${workflowDir}. Use --force to overwrite.\n`,
);
process.exit(1);
}
mkdirSync(workflowDir, { recursive: true });
writeFile(join(workflowDir, "index.ts"), buildWorkflowTemplate(args.name));
process.stdout.write(`✅ Workflow scaffolded: ${workflowDir}/index.ts\n`);
process.stdout.write("\n💡 Next steps:\n");
process.stdout.write(" 1. Add to nerve.yaml:\n");
process.stdout.write(" workflows:\n");
process.stdout.write(` ${args.name}:\n`);
process.stdout.write(" concurrency: 1\n");
process.stdout.write(" overflow: drop\n");
process.stdout.write(` 2. Edit ${workflowDir}/index.ts to implement your roles.\n`);
process.stdout.write(" 3. Run `nerve start` to launch the daemon.\n");
},
});
const initWorkspaceCommand = defineCommand({
meta: {
name: "workspace",
description: "Initialize the ~/.uncaged-nerve/ workspace (default)",
},
args: {
force: {
@@ -127,44 +215,115 @@ export const initCommand = defineCommand({
},
},
async run({ args }) {
const nerveRoot = getNerveRoot();
if (existsSync(nerveRoot) && !args.force) {
process.stderr.write("⚠️ ~/.uncaged-nerve/ already exists. Use --force to reinitialize.\n");
process.exit(1);
}
mkdirSync(join(nerveRoot, "data"), { recursive: true });
mkdirSync(join(nerveRoot, "senses", "cpu-usage", "migrations"), { recursive: true });
writeFile(join(nerveRoot, "nerve.yaml"), NERVE_YAML);
writeFile(join(nerveRoot, "package.json"), PACKAGE_JSON);
writeFile(join(nerveRoot, ".gitignore"), GITIGNORE);
writeFile(join(nerveRoot, "senses", "cpu-usage", "schema.ts"), CPU_SCHEMA_TS);
writeFile(join(nerveRoot, "senses", "cpu-usage", "index.ts"), CPU_INDEX_TS);
writeFile(
join(nerveRoot, "senses", "cpu-usage", "migrations", "0001_init.sql"),
CPU_MIGRATION_SQL,
);
process.stdout.write("Installing dependencies…\n");
try {
const { cmd, args } = await detectPackageManager();
await runCommand(cmd, args, nerveRoot);
} catch {
process.stdout.write("⚠️ Install failed — you may need to install dependencies manually.\n");
}
if (!existsSync(join(nerveRoot, ".git"))) {
try {
await runCommand("git", ["init"], nerveRoot);
} catch {
process.stdout.write("⚠️ git init failed — skipping.\n");
}
}
process.stdout.write(
"✅ Workspace created at ~/.uncaged-nerve/\n 1 example sense: cpu-usage\n Run `nerve start` to launch the daemon.\n",
);
await runInitWorkspace(args.force);
},
});
async function tryRequireSqlite(nerveRoot: string): Promise<boolean> {
try {
const modulePath = join(nerveRoot, "node_modules", "better-sqlite3");
// Use a child process to test if the native module loads
const { execFile } = await import("node:child_process");
const { promisify } = await import("node:util");
const execFileAsync = promisify(execFile);
await execFileAsync("node", ["-e", `require(${JSON.stringify(modulePath)})`], {
cwd: nerveRoot,
timeout: 10_000,
});
return true;
} catch {
return false;
}
}
async function runInitWorkspace(force: boolean): Promise<void> {
const nerveRoot = getNerveRoot();
if (existsSync(nerveRoot) && !force) {
process.stderr.write("⚠️ ~/.uncaged-nerve/ already exists. Use --force to reinitialize.\n");
process.exit(1);
}
mkdirSync(join(nerveRoot, "data"), { recursive: true });
mkdirSync(join(nerveRoot, "data", "senses"), { recursive: true });
mkdirSync(join(nerveRoot, "senses", "cpu-usage", "migrations"), { recursive: true });
writeFile(join(nerveRoot, "nerve.yaml"), NERVE_YAML);
writeFile(join(nerveRoot, "package.json"), PACKAGE_JSON);
writeFile(join(nerveRoot, ".gitignore"), GITIGNORE);
writeFile(join(nerveRoot, "senses", "cpu-usage", "schema.ts"), CPU_SCHEMA_TS);
writeFile(join(nerveRoot, "senses", "cpu-usage", "index.js"), CPU_INDEX_JS);
writeFile(
join(nerveRoot, "senses", "cpu-usage", "migrations", "0001_init.sql"),
CPU_MIGRATION_SQL,
);
process.stdout.write("Installing dependencies…\n");
const { cmd, installArgs } = await detectPackageManager();
try {
await runCommand(cmd, installArgs, nerveRoot);
} catch {
process.stdout.write(
`⚠️ Install failed. Try manually:\n cd ${nerveRoot} && ${cmd} ${installArgs.join(" ")}\n`,
);
}
// Verify better-sqlite3 native module — rebuild up to 2 times if broken
const sqlitePath = join(nerveRoot, "node_modules", "better-sqlite3");
if (existsSync(sqlitePath)) {
for (let attempt = 1; attempt <= 2; attempt++) {
if (await tryRequireSqlite(nerveRoot)) break;
process.stdout.write(
`${attempt === 1 ? "Building" : "Retrying build of"} native module better-sqlite3 (attempt ${attempt}/2)…\n`,
);
try {
await runCommand(cmd, ["rebuild", "better-sqlite3"], nerveRoot);
} catch {
// will be caught by the verify below
}
}
if (!(await tryRequireSqlite(nerveRoot))) {
process.stdout.write(
`⚠️ better-sqlite3 native module is not working. The daemon will fail to start.\n` +
` Fix: cd ${nerveRoot} && ${cmd} rebuild better-sqlite3\n` +
` Or: npm install --build-from-source better-sqlite3\n`,
);
}
}
if (!existsSync(join(nerveRoot, ".git"))) {
try {
await runCommand("git", ["init"], nerveRoot);
await runCommand("git", ["add", "."], nerveRoot);
await runCommand("git", ["commit", "-m", "Initial nerve workspace"], nerveRoot);
} catch {
process.stdout.write("⚠️ git init failed — skipping.\n");
}
}
process.stdout.write(
"✅ Workspace created at ~/.uncaged-nerve/\n 1 example sense: cpu-usage\n Run `nerve start` to launch the daemon.\n",
);
}
export const initCommand = defineCommand({
meta: {
name: "init",
description:
"Initialize workspace (nerve init) or scaffold templates (nerve init workflow <name>)",
},
args: {
force: {
type: "boolean",
description: "Reinitialize even if workspace already exists (preserves data/)",
default: false,
},
},
subCommands: {
workflow: initWorkflowCommand,
workspace: initWorkspaceCommand,
},
async run({ args }) {
await runInitWorkspace(args.force);
},
});
+197
View File
@@ -0,0 +1,197 @@
import { createReadStream, existsSync, statSync } from "node:fs";
import { createInterface } from "node:readline";
import { defineCommand } from "citty";
import { getLogPath } from "../workspace.js";
export const DEFAULT_LOG_LINES = 50;
const sleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
/**
* Read all lines from a file. Returns empty array if file does not exist.
*
* TODO: For tail mode (offset=0), avoid reading the whole file into memory by
* seeking to the last N bytes via createReadStream({ start: max(0, size - CHUNK) }).
*/
export async function readAllLines(filePath: string): Promise<string[]> {
if (!existsSync(filePath)) return [];
const lines: string[] = [];
const rl = createInterface({
input: createReadStream(filePath, { encoding: "utf8" }),
crlfDelay: Number.POSITIVE_INFINITY,
});
for await (const line of rl) {
lines.push(line);
}
return lines;
}
/**
* Slice a log line array respecting offset + limit semantics.
*
* When offset is 0 the function returns the *last* `limit` lines (tail mode).
* When offset > 0 it is treated as a 1-based line number and the slice starts
* there (for pagination of earlier pages from the tail).
*
* Returns the selected lines plus metadata used to build the footer.
*/
export type LogSlice = {
lines: string[];
total: number;
startLine: number; // 1-based, inclusive
endLine: number; // 1-based, inclusive
nextOffset: number | null; // null when no previous page exists
};
export function sliceLogs(allLines: string[], offset: number, limit: number): LogSlice {
const total = allLines.length;
if (total === 0) {
return { lines: [], total: 0, startLine: 0, endLine: 0, nextOffset: null };
}
let start: number;
if (offset === 0) {
// Tail mode: last `limit` lines
start = Math.max(0, total - limit);
} else {
// offset is 1-based line number
start = Math.max(0, offset - 1);
}
const end = Math.min(start + limit, total);
const lines = allLines.slice(start, end);
const startLine = start + 1;
const endLine = end;
// nextOffset points to lines *before* current slice (earlier in file)
const nextOffset = start > 0 ? Math.max(1, startLine - limit) : null;
return { lines, total, startLine, endLine, nextOffset };
}
/**
* Build the footer string shown after the log lines.
*/
export function buildLogFooter(slice: LogSlice, nArg: number, logPath: string): string {
if (slice.total === 0) {
return "📭 Log file is empty.\n";
}
const rangeStr = `lines ${slice.startLine}-${slice.endLine} of ${slice.total}`;
let footer = `\n📄 ${rangeStr} | ${logPath}\n`;
if (slice.nextOffset !== null) {
footer += `⏩ Earlier lines available. Fetch previous page:\n`;
footer += ` nerve logs --offset ${slice.nextOffset} -n ${nArg}\n`;
}
return footer;
}
// ---------------------------------------------------------------------------
// nerve logs
// ---------------------------------------------------------------------------
export const logsCommand = defineCommand({
meta: {
name: "logs",
description: "Show daemon log output",
},
args: {
n: {
type: "string",
description: `Number of lines to show (default: ${DEFAULT_LOG_LINES})`,
default: String(DEFAULT_LOG_LINES),
},
offset: {
type: "string",
description: "Start from line N (1-based, for pagination)",
default: "0",
},
follow: {
type: "boolean",
alias: "f",
description: "Stream new log lines in real time",
default: false,
},
},
async run({ args }) {
const logPath = getLogPath();
const nLines = Math.max(1, Number.parseInt(args.n, 10) || DEFAULT_LOG_LINES);
const rawOffset = Number.parseInt(args.offset, 10) || 0;
if (rawOffset < 0) {
process.stderr.write(`❌ --offset must be a non-negative integer, got: ${args.offset}\n`);
process.exit(1);
}
const offset = rawOffset;
if (!existsSync(logPath)) {
process.stderr.write(`❌ Log file not found: ${logPath}\n`);
process.stderr.write(" Has the daemon been started? Try: nerve start\n");
process.exit(1);
}
if (args.follow) {
await followLog(logPath, nLines);
return;
}
const allLines = await readAllLines(logPath);
const slice = sliceLogs(allLines, offset, nLines);
for (const line of slice.lines) {
process.stdout.write(`${line}\n`);
}
process.stdout.write(buildLogFooter(slice, nLines, logPath));
},
});
/**
* Stream new lines from a log file as they are appended.
* Shows the last `tailLines` lines first, then watches for new content.
*/
async function followLog(logPath: string, tailLines: number): Promise<void> {
const allLines = await readAllLines(logPath);
const initial = allLines.slice(Math.max(0, allLines.length - tailLines));
for (const line of initial) {
process.stdout.write(`${line}\n`);
}
let size = statSync(logPath).size;
process.stdout.write(`\n👁 Following ${logPath} — press Ctrl+C to stop\n`);
let stopped = false;
process.once("SIGINT", () => {
stopped = true;
});
while (!stopped) {
await sleep(300);
if (stopped) break;
try {
const newSize = statSync(logPath).size;
if (newSize < size) {
// Log rotation: file was truncated or replaced, read from the beginning
size = 0;
}
if (newSize <= size) continue;
const stream = createReadStream(logPath, { start: size, encoding: "utf8" });
const rl = createInterface({ input: stream, crlfDelay: Number.POSITIVE_INFINITY });
for await (const line of rl) {
process.stdout.write(`${line}\n`);
}
size = newSize;
} catch {
stopped = true;
}
}
}
+155
View File
@@ -0,0 +1,155 @@
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { type SenseInfo, parseNerveConfig } from "@uncaged/nerve-core";
import { defineCommand } from "citty";
import { listSensesViaDaemon, triggerSenseViaDaemon } from "../daemon-client.js";
import { getNerveRoot, getSocketPath, isRunning } from "../workspace.js";
// ---------------------------------------------------------------------------
// Formatting helpers (exported for tests)
// ---------------------------------------------------------------------------
export function formatDuration(ms: number | null): string {
if (ms === null) return "—";
const totalSeconds = Math.floor(ms / 1000);
if (totalSeconds < 60) return `${totalSeconds}s`;
const minutes = Math.floor(totalSeconds / 60);
const seconds = totalSeconds % 60;
if (minutes < 60) return `${minutes}m ${seconds}s`;
const hours = Math.floor(minutes / 60);
const remainingMinutes = minutes % 60;
return `${hours}h ${remainingMinutes}m`;
}
export function formatSenseList(senses: SenseInfo[]): string {
if (senses.length === 0) {
return "📭 No senses registered in nerve.yaml.\n";
}
const lines: string[] = [`📡 Registered senses (${senses.length}):\n`];
for (const s of senses) {
lines.push(`\n ${s.name}\n`);
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)";
lines.push(` last signal: ${lastSignal}\n`);
}
return lines.join("");
}
/** Build a SenseInfo list from nerve.yaml when daemon is not running. */
export function sensesFromConfig(configPath: string): SenseInfo[] {
let raw: string;
try {
raw = readFileSync(configPath, "utf8");
} catch {
return [];
}
const result = parseNerveConfig(raw);
if (!result.ok) return [];
return Object.entries(result.value.senses).map(([name, cfg]) => ({
name,
group: cfg.group,
throttle: cfg.throttle,
timeout: cfg.timeout,
lastSignalTs: null,
}));
}
// ---------------------------------------------------------------------------
// nerve sense list
// ---------------------------------------------------------------------------
const senseListCommand = defineCommand({
meta: {
name: "list",
description: "List all registered senses and their status",
},
async run() {
if (!isRunning()) {
// Daemon not running — show static info from nerve.yaml
process.stderr.write(
"⚠️ Daemon is not running — showing static config only (no last signal time).\n\n",
);
const configPath = join(getNerveRoot(), "nerve.yaml");
const senses = sensesFromConfig(configPath);
process.stdout.write(formatSenseList(senses));
return;
}
const socketPath = getSocketPath();
let response: { ok: true; senses: SenseInfo[] } | { ok: false; error: string };
try {
response = await listSensesViaDaemon(socketPath);
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`❌ Failed to contact daemon: ${msg}\n`);
process.exit(1);
}
if (!response.ok) {
process.stderr.write(`❌ Daemon error: ${response.error}\n`);
process.exit(1);
}
process.stdout.write(formatSenseList(response.senses));
},
});
// ---------------------------------------------------------------------------
// nerve sense trigger <name>
// ---------------------------------------------------------------------------
const senseTriggerCommand = defineCommand({
meta: {
name: "trigger",
description: "Manually trigger a sense compute by sending an IPC message to the running daemon",
},
args: {
name: {
type: "positional",
description: "The sense name to trigger",
},
},
async run({ args }) {
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 };
try {
response = await triggerSenseViaDaemon(socketPath, args.name);
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`❌ Failed to contact daemon: ${msg}\n`);
process.exit(1);
}
if (!response.ok) {
process.stderr.write(`❌ Daemon rejected trigger: ${response.error}\n`);
process.exit(1);
}
process.stdout.write(`✅ Triggered sense "${args.name}" via daemon.\n`);
},
});
// ---------------------------------------------------------------------------
// nerve sense (parent command)
// ---------------------------------------------------------------------------
export const senseCommand = defineCommand({
meta: {
name: "sense",
description: "Interact with sense computes",
},
subCommands: {
list: senseListCommand,
trigger: senseTriggerCommand,
},
});
+69 -92
View File
@@ -1,78 +1,57 @@
import { createWriteStream } from "node:fs";
import { readFileSync } from "node:fs";
import { createWriteStream, existsSync } from "node:fs";
import { mkdir } from "node:fs/promises";
import { join } from "node:path";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import { parseNerveConfig } from "@uncaged/nerve-core";
import { createKernel } from "@uncaged/nerve-daemon";
import { defineCommand } from "citty";
import { getLogPath, getNerveRoot, isRunning, readPidFile, writePidFile } from "../workspace.js";
import {
getLogPath,
getNerveRoot,
isRunning,
readPidFile,
removePidFile,
writePidFile,
} from "../workspace.js";
function readConfig(nerveRoot: string): ReturnType<typeof parseNerveConfig> {
const configPath = join(nerveRoot, "nerve.yaml");
let raw: string;
try {
raw = readFileSync(configPath, "utf8");
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
return { ok: false, error: new Error(`❌ Cannot read ${configPath}: ${msg}`) };
}
return parseNerveConfig(raw);
function waitForSocket(socketPath: string, timeoutMs = 5000, intervalMs = 200): Promise<boolean> {
return new Promise((resolve) => {
const deadline = Date.now() + timeoutMs;
const check = (): void => {
if (existsSync(socketPath)) {
resolve(true);
} else if (Date.now() >= deadline) {
resolve(false);
} else {
setTimeout(check, intervalMs);
}
};
check();
});
}
async function runForeground(nerveRoot: string): Promise<void> {
const configResult = readConfig(nerveRoot);
if (!configResult.ok) {
process.stderr.write(`${configResult.error.message}\n`);
process.exit(1);
/** Path to the CLI entry script (used to locate dist/ next to bundled assets). */
function cliEntryScript(): string {
const here = fileURLToPath(import.meta.url);
const ext = here.endsWith(".ts") ? ".ts" : ".js";
const candidates = [join(dirname(here), `cli${ext}`), join(dirname(here), "..", `cli${ext}`)];
const cliPath = candidates.find((p) => existsSync(p));
if (!cliPath) {
throw new Error(`CLI entry not found (searched: ${candidates.join(", ")})`);
}
return cliPath;
}
const config = configResult.value;
const kernel = createKernel(config, nerveRoot);
const senseNames = Object.keys(config.senses);
const groups = [...kernel.groups];
process.stdout.write(
`✅ Nerve starting — ${senseNames.length} sense(s), ${groups.length} group(s)\n`,
function daemonBootstrapScript(): string {
const cliPath = cliEntryScript();
const dir = dirname(cliPath);
const bootstrapJs = join(dir, "daemon-bootstrap.js");
if (existsSync(bootstrapJs)) {
return bootstrapJs;
}
throw new Error(
`daemon-bootstrap.js not found next to CLI at ${bootstrapJs}. Build the CLI package (e.g. \`pnpm --filter @uncaged/nerve-cli build\`) before using \`nerve daemon start\`.`,
);
for (const group of groups) {
const groupSenses = Object.entries(config.senses)
.filter(([, sc]) => sc.group === group)
.map(([name]) => name);
process.stdout.write(` group "${group}": ${groupSenses.join(", ")}\n`);
}
process.stdout.write(" Press Ctrl+C to stop.\n");
let shuttingDown = false;
async function shutdown(): Promise<void> {
if (shuttingDown) return;
shuttingDown = true;
process.stdout.write("\n[nerve] Shutting down…\n");
await kernel.stop();
process.exit(0);
}
process.on("SIGINT", () => {
shutdown().catch((e: unknown) => {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`[nerve] Shutdown error: ${msg}\n`);
process.exit(1);
});
});
process.on("SIGTERM", () => {
shutdown().catch((e: unknown) => {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`[nerve] Shutdown error: ${msg}\n`);
process.exit(1);
});
});
await kernel.ready;
}
async function runDaemon(nerveRoot: string): Promise<void> {
@@ -82,12 +61,6 @@ async function runDaemon(nerveRoot: string): Promise<void> {
process.exit(1);
}
const configResult = readConfig(nerveRoot);
if (!configResult.ok) {
process.stderr.write(`${configResult.error.message}\n`);
process.exit(1);
}
const logPath = getLogPath();
await mkdir(join(nerveRoot, "logs"), { recursive: true });
@@ -98,12 +71,13 @@ async function runDaemon(nerveRoot: string): Promise<void> {
else resolve();
});
const selfPath = fileURLToPath(import.meta.url);
const bootstrapPath = daemonBootstrapScript();
const child = spawn(process.execPath, [selfPath, "start"], {
const child = spawn(process.execPath, [bootstrapPath], {
detached: true,
stdio: ["ignore", logStream.fd, logStream.fd],
env: { ...process.env, NERVE_DAEMON_MODE: "1" },
env: { ...process.env, NERVE_ROOT: nerveRoot },
cwd: nerveRoot,
});
child.unref();
@@ -115,31 +89,34 @@ async function runDaemon(nerveRoot: string): Promise<void> {
}
writePidFile(pid);
const { getSocketPath } = await import("../workspace.js");
const ready = await waitForSocket(getSocketPath(), 5000);
if (!ready || !isRunning()) {
removePidFile();
process.stderr.write(
`❌ Daemon process exited shortly after start. Check logs at:\n ${logPath}\n`,
);
process.exit(1);
}
process.stdout.write(`✅ Nerve daemon started (pid ${pid}).\n`);
process.stdout.write(` Logs: ${logPath}\n`);
process.stdout.write(" Run `nerve stop` to stop.\n");
process.stdout.write(" Run `nerve daemon stop` (or `nerve stop`) to stop.\n");
}
export const startCommand = defineCommand({
/** Background daemon only — use `nerve dev` for foreground mode. */
export async function runDaemonStartCommand(): Promise<void> {
await runDaemon(getNerveRoot());
}
export const daemonStartCommand = defineCommand({
meta: {
name: "start",
description: "Start the nerve daemon",
description: "Start the nerve daemon in the background",
},
args: {
daemon: {
type: "boolean",
alias: "d",
description: "Run as background daemon",
default: false,
},
},
async run({ args }) {
const nerveRoot = getNerveRoot();
if (args.daemon) {
await runDaemon(nerveRoot);
} else {
await runForeground(nerveRoot);
}
async run() {
await runDaemonStartCommand();
},
});
+38 -33
View File
@@ -15,44 +15,49 @@ async function waitForExit(pid: number, timeoutMs: number): Promise<boolean> {
return false;
}
/** Core stop logic — also used by `nerve daemon restart`. */
export async function runStopCommand(): Promise<void> {
const pid = readPidFile();
if (pid === null) {
process.stdout.write("⚠️ No PID file found — daemon may not be running.\n");
return;
}
if (!isRunning()) {
process.stdout.write("⚠️ Daemon is not running (stale PID file). Cleaning up.\n");
removePidFile();
return;
}
process.stdout.write(`Stopping nerve daemon (pid ${pid})…\n`);
try {
process.kill(pid, "SIGTERM");
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`❌ Failed to send SIGTERM: ${msg}\n`);
process.exit(1);
}
const graceful = await waitForExit(pid, 10_000);
if (!graceful) {
process.stdout.write("⚠️ Daemon did not exit in 10s — sending SIGKILL.\n");
try {
process.kill(pid, "SIGKILL");
} catch {
// already dead
}
}
removePidFile();
process.stdout.write("✅ Daemon stopped.\n");
}
export const stopCommand = defineCommand({
meta: {
name: "stop",
description: "Stop the nerve daemon",
},
async run() {
const pid = readPidFile();
if (pid === null) {
process.stdout.write("⚠️ No PID file found — daemon may not be running.\n");
return;
}
if (!isRunning()) {
process.stdout.write("⚠️ Daemon is not running (stale PID file). Cleaning up.\n");
removePidFile();
return;
}
process.stdout.write(`Stopping nerve daemon (pid ${pid})…\n`);
try {
process.kill(pid, "SIGTERM");
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`❌ Failed to send SIGTERM: ${msg}\n`);
process.exit(1);
}
const graceful = await waitForExit(pid, 10_000);
if (!graceful) {
process.stdout.write("⚠️ Daemon did not exit in 10s — sending SIGKILL.\n");
try {
process.kill(pid, "SIGKILL");
} catch {
// already dead
}
}
removePidFile();
process.stdout.write("✅ Daemon stopped.\n");
await runStopCommand();
},
});
+70
View File
@@ -0,0 +1,70 @@
import { existsSync } from "node:fs";
import { join } from "node:path";
import { defineCommand } from "citty";
import { loadDaemonModule } from "../workspace-daemon.js";
import { getNerveRoot } from "../workspace.js";
// ---------------------------------------------------------------------------
// nerve store archive
// ---------------------------------------------------------------------------
const storeArchiveCommand = defineCommand({
meta: {
name: "archive",
description:
"Export logs older than 30 days from logs.db to data/archive/logs/YYYY-MM-DD.jsonl and delete those rows (RFC-001 §5.4)",
},
args: {
vacuum: {
type: "boolean",
description: "Run SQLite VACUUM after archiving",
default: false,
},
},
async run({ args }) {
const nerveRoot = getNerveRoot();
const dbPath = join(nerveRoot, "data", "logs.db");
if (!existsSync(dbPath)) {
process.stderr.write("❌ No data/logs.db found — start the daemon at least once.\n");
process.exit(1);
}
const { createLogStore } = await loadDaemonModule(nerveRoot);
const store = createLogStore(dbPath);
try {
const result = store.archiveLogs({ vacuum: args.vacuum });
if (result.days.length === 0) {
process.stdout.write(
"✅ Nothing to archive (no eligible UTC days beyond the 30-day window).\n",
);
} else {
process.stdout.write(`✅ Archived ${result.days.length} day(s):\n`);
for (const d of result.days) {
process.stdout.write(` ${d.day} rows=${d.rowCount} ${d.filePath}\n`);
}
}
if (result.vacuumed) {
process.stdout.write(" VACUUM completed.\n");
}
} finally {
store.close();
}
},
});
// ---------------------------------------------------------------------------
// nerve store
// ---------------------------------------------------------------------------
export const storeCommand = defineCommand({
meta: {
name: "store",
description: "Maintain local Nerve SQLite stores (log cold-archive, …)",
},
subCommands: {
archive: storeArchiveCommand,
},
});
+364
View File
@@ -0,0 +1,364 @@
import { existsSync } from "node:fs";
import { join } from "node:path";
import { defineCommand } from "citty";
import { triggerWorkflowViaDaemon } from "../daemon-client.js";
import type { LogStore, WorkflowRun } from "../daemon-types.js";
import { loadDaemonModule } from "../workspace-daemon.js";
import { getNerveRoot, getSocketPath, isRunning } from "../workspace.js";
export const DEFAULT_PAGE_SIZE = 20;
export function parseIntArg(raw: string, fallback: number): number {
const v = Number.parseInt(raw, 10);
return Number.isNaN(v) ? fallback : v;
}
export function getDbPath(): string {
return join(getNerveRoot(), "data", "logs.db");
}
export function formatTs(ts: number): string {
return new Date(ts).toISOString();
}
async function openStore(): Promise<LogStore> {
const nerveRoot = getNerveRoot();
const dbPath = getDbPath();
if (!existsSync(dbPath)) {
process.stderr.write("❌ No logs.db found — has the daemon run yet?\n");
process.exit(1);
}
const { createLogStore } = await loadDaemonModule(nerveRoot);
return createLogStore(dbPath);
}
export function statusIcon(status: WorkflowRun["status"]): string {
switch (status) {
case "started":
return "▶";
case "queued":
return "⏳";
case "completed":
return "✅";
case "failed":
return "❌";
case "crashed":
return "💥";
case "dropped":
return "🗑";
case "interrupted":
return "⚠️";
default: {
const _exhaustive: never = status;
return `?(${_exhaustive})`;
}
}
}
/**
* Retrieve all workflow runs from the store, sorted by ts descending (newest first).
* Delegates to the store's efficient SQL query on the workflow_runs table.
*/
export function getAllWorkflowRuns(store: LogStore, filterWorkflow: string | null): WorkflowRun[] {
return store.getAllWorkflowRuns(filterWorkflow);
}
/**
* Format a single workflow run as a single output line (no trailing newline in icon/fields).
*/
export function formatRunLine(run: WorkflowRun): string {
const icon = statusIcon(run.status);
return ` ${icon} ${run.runId} workflow=${run.workflow} status=${run.status} ts=${formatTs(run.ts)}\n`;
}
/**
* Format a paginated list of workflow runs into output lines.
* Returns the lines to write and any pagination hint.
*/
export type ListOutput = {
lines: string[];
paginationHint: string | null;
};
export function buildListOutput(
runs: WorkflowRun[],
offset: number,
limit: number,
allFlag: boolean,
filterWorkflow: string | null,
): ListOutput {
const total = runs.length;
const page = runs.slice(offset, offset + limit);
const shown = page.length;
const remaining = total - offset - shown;
if (total === 0) {
const msg = allFlag
? "📭 No workflow runs found.\n"
: "📭 No active workflow runs. Use --all to include completed/failed runs.\n";
return { lines: [msg], paginationHint: null };
}
const lines: string[] = [];
lines.push(`📋 Workflow runs (${shown} of ${total} shown):\n`);
for (const run of page) {
lines.push(formatRunLine(run));
}
let paginationHint: string | null = null;
if (remaining > 0) {
const wfFlag = filterWorkflow !== null ? ` --workflow ${filterWorkflow}` : "";
const allFlagStr = allFlag ? " --all" : "";
paginationHint =
`\n⏩ ${remaining} more run(s) not shown. Fetch next page:\n` +
` nerve workflow list --offset ${offset + limit}${allFlagStr}${wfFlag}\n`;
}
return { lines, paginationHint };
}
/**
* Format the inspect output for a single run's log entries with pagination.
*/
export type InspectOutput = {
header: string[];
eventLines: string[];
paginationHint: string | null;
};
export function buildInspectOutput(
run: WorkflowRun,
allLogs: Array<{ ts: number; type: string; payload: string | null }>,
offset: number,
limit: number,
): InspectOutput {
const total = allLogs.length;
const page = allLogs.slice(offset, offset + limit);
const shown = page.length;
const remaining = total - offset - shown;
const header: string[] = [
`🔍 Workflow run: ${run.runId}\n`,
` workflow: ${run.workflow}\n`,
` status: ${run.status}\n`,
` ts: ${formatTs(run.ts)}\n`,
`\n📜 Thread events (${shown} of ${total}):\n`,
];
const eventLines: string[] = [];
if (total === 0) {
eventLines.push(" (no events recorded)\n");
} else {
for (const entry of page) {
const payloadStr =
entry.payload === null
? ""
: entry.payload.length <= 200
? ` payload=${entry.payload}`
: ` payload=${entry.payload.slice(0, 200)}`;
eventLines.push(` [${formatTs(entry.ts)}] type=${entry.type}${payloadStr}\n`);
}
}
let paginationHint: string | null = null;
if (remaining > 0) {
paginationHint =
`\n⏩ ${remaining} more event(s) not shown. Fetch next page:\n` +
` nerve workflow inspect ${run.runId} --offset ${offset + limit}\n`;
}
return { header, eventLines, paginationHint };
}
// ---------------------------------------------------------------------------
// nerve workflow list
// ---------------------------------------------------------------------------
const workflowListCommand = defineCommand({
meta: {
name: "list",
description: "List active (queued/started) workflow runs",
},
args: {
all: {
type: "boolean",
description: "Include completed/failed/crashed runs",
default: false,
},
workflow: {
type: "string",
description: "Filter by workflow name",
default: "",
},
limit: {
type: "string",
description: `Max runs to show (default: ${DEFAULT_PAGE_SIZE})`,
default: String(DEFAULT_PAGE_SIZE),
},
offset: {
type: "string",
description: "Skip first N runs (for pagination)",
default: "0",
},
},
async run({ args }) {
const store = await openStore();
try {
const limit = Math.max(1, parseIntArg(args.limit, DEFAULT_PAGE_SIZE));
const offset = Math.max(0, parseIntArg(args.offset, 0));
const filterWorkflow = args.workflow.length > 0 ? args.workflow : null;
const runs = args.all
? getAllWorkflowRuns(store, filterWorkflow)
: store.getActiveWorkflowRuns(filterWorkflow ?? undefined);
const { lines, paginationHint } = buildListOutput(
runs,
offset,
limit,
args.all,
filterWorkflow,
);
for (const line of lines) {
process.stdout.write(line);
}
if (paginationHint !== null) {
process.stdout.write(paginationHint);
}
} finally {
store.close();
}
},
});
// ---------------------------------------------------------------------------
// nerve workflow inspect <runId>
// ---------------------------------------------------------------------------
const workflowInspectCommand = defineCommand({
meta: {
name: "inspect",
description: "Show details and thread events for a workflow run",
},
args: {
runId: {
type: "positional",
description: "The run ID to inspect",
},
limit: {
type: "string",
description: `Max log entries to show (default: ${DEFAULT_PAGE_SIZE})`,
default: String(DEFAULT_PAGE_SIZE),
},
offset: {
type: "string",
description: "Skip first N log entries (for pagination)",
default: "0",
},
},
async run({ args }) {
const store = await openStore();
try {
const limit = Math.max(1, parseIntArg(args.limit, DEFAULT_PAGE_SIZE));
const offset = Math.max(0, parseIntArg(args.offset, 0));
const run = store.getWorkflowRun(args.runId);
if (run === null) {
process.stderr.write(`❌ No workflow run found with runId: ${args.runId}\n`);
process.exit(1);
}
const allLogs = store.query({ source: "workflow", refId: args.runId });
const { header, eventLines, paginationHint } = buildInspectOutput(
run,
allLogs,
offset,
limit,
);
for (const line of [...header, ...eventLines]) {
process.stdout.write(line);
}
if (paginationHint !== null) {
process.stdout.write(paginationHint);
}
} finally {
store.close();
}
},
});
// ---------------------------------------------------------------------------
// nerve workflow trigger <name>
// ---------------------------------------------------------------------------
const workflowTriggerCommand = defineCommand({
meta: {
name: "trigger",
description: "Manually trigger a workflow by sending an IPC message to the running daemon",
},
args: {
name: {
type: "positional",
description: "The workflow name to trigger",
},
payload: {
type: "string",
description: "JSON payload to pass as trigger payload (default: {})",
default: "{}",
},
},
async run({ args }) {
let triggerPayload: unknown = {};
try {
triggerPayload = JSON.parse(args.payload) as unknown;
} catch {
process.stderr.write(`❌ --payload must be valid JSON. Got: ${args.payload}\n`);
process.exit(1);
}
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 };
try {
response = await triggerWorkflowViaDaemon(socketPath, args.name, triggerPayload);
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`❌ Failed to contact daemon: ${msg}\n`);
process.exit(1);
}
if (!response.ok) {
process.stderr.write(`❌ Daemon rejected trigger: ${response.error}\n`);
process.exit(1);
}
process.stdout.write(`✅ Triggered workflow "${args.name}" via daemon.\n`);
process.stdout.write("\n💡 Inspect active runs with: nerve workflow list\n");
},
});
// ---------------------------------------------------------------------------
// nerve workflow (parent command)
// ---------------------------------------------------------------------------
export const workflowCommand = defineCommand({
meta: {
name: "workflow",
description: "Manage and inspect workflow runs",
},
subCommands: {
list: workflowListCommand,
inspect: workflowInspectCommand,
trigger: workflowTriggerCommand,
},
});
+11
View File
@@ -0,0 +1,11 @@
import { runForegroundKernelSession } from "./run-foreground-kernel.js";
import { loadDaemonModule } from "./workspace-daemon.js";
const nerveRoot = process.env.NERVE_ROOT;
if (nerveRoot === undefined || nerveRoot.length === 0) {
process.stderr.write("[nerve] NERVE_ROOT environment variable is required.\n");
process.exit(1);
}
const { createKernel } = await loadDaemonModule(nerveRoot);
await runForegroundKernelSession(nerveRoot, createKernel);
+145
View File
@@ -0,0 +1,145 @@
/**
* Daemon IPC client — connects to the daemon's Unix socket and sends
* trigger-workflow or trigger-sense requests.
*
* Protocol: newline-delimited JSON (same as daemon-ipc.ts server side).
*/
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;
export type { SenseInfo };
type TriggerResponse = { ok: true } | { ok: false; error: string };
type ListSensesResponse = { ok: true; senses: SenseInfo[] } | { ok: false; error: string };
function parseDaemonResponse(line: string): TriggerResponse {
try {
const obj = JSON.parse(line) as unknown;
if (obj !== null && typeof obj === "object") {
const r = obj as Record<string, unknown>;
if (r.ok === true) return { ok: true };
if (r.ok === false && typeof r.error === "string") return { ok: false, error: r.error };
}
} catch {
// fall through
}
return { ok: false, error: `Unexpected daemon response: ${line}` };
}
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: T | 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"));
}, responseTimeoutMs);
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);
settle(parseFirstLine(trimmed));
return;
}
});
const msg = `${JSON.stringify(message)}\n`;
socket?.write(msg);
});
socket.on("error", (err) => {
clearTimeout(connectTimer);
settle(new Error(`Cannot connect to daemon: ${err.message}`));
});
});
}
/**
* Send a trigger-workflow message to the running daemon via its Unix socket.
* Resolves with the daemon's response or rejects on connection/timeout errors.
*/
export function triggerWorkflowViaDaemon(
socketPath: string,
workflow: string,
payload: unknown,
): Promise<TriggerResponse> {
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 }, 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);
}
+70
View File
@@ -0,0 +1,70 @@
/**
* Structural types for workflow CLI — mirrors @uncaged/nerve-daemon log-store
* public API so the CLI runtime does not statically depend on the daemon package.
*
* ⚠️ Keep in sync with @uncaged/nerve-daemon exports.
* Run `pnpm --filter @uncaged/nerve-cli test` to catch drift via satisfies assertions.
*/
export type WorkflowRunStatus =
| "queued"
| "started"
| "completed"
| "failed"
| "crashed"
| "dropped"
| "interrupted";
export type WorkflowRun = {
runId: string;
workflow: string;
status: WorkflowRunStatus;
ts: number;
};
export type LogEntry = {
id?: number;
source: string;
type: string;
refId: string | null;
payload: string | null;
ts: number;
};
export type LogQuery = {
source?: string;
type?: string;
refId?: string;
since?: number;
until?: number;
limit?: number;
};
export type ArchiveLogsOptions = {
now?: number;
vacuum?: boolean;
maxDays?: number;
retentionMs?: number;
};
export type ArchiveLogsDayResult = {
day: string;
rowCount: number;
filePath: string;
};
export type ArchiveLogsResult = {
days: ArchiveLogsDayResult[];
vacuumed: boolean;
};
/** Subset of daemon LogStore used by the CLI workflow commands. */
export type LogStore = {
query: (filter?: LogQuery) => LogEntry[];
getWorkflowRun: (runId: string) => WorkflowRun | null;
getActiveWorkflowRuns: (workflowName?: string) => WorkflowRun[];
getAllWorkflowRuns: (workflowName: string | null) => WorkflowRun[];
upsertWorkflowRun: (entry: Omit<LogEntry, "id">, run: WorkflowRun) => LogEntry;
archiveLogs: (options?: ArchiveLogsOptions) => ArchiveLogsResult;
close: () => void;
};
+5 -2
View File
@@ -8,5 +8,8 @@ export {
isRunning,
} from "./workspace.js";
export { createKernel } from "@uncaged/nerve-daemon";
export type { Kernel } from "@uncaged/nerve-daemon";
export {
assertWorkspaceDaemonInstalled,
getDaemonEntryPath,
loadDaemonModule,
} from "./workspace-daemon.js";
+88
View File
@@ -0,0 +1,88 @@
import { readFileSync } from "node:fs";
import { join } from "node:path";
import type { NerveConfig } from "@uncaged/nerve-core";
import { parseNerveConfig } from "@uncaged/nerve-core";
import { getSocketPath } from "./workspace.js";
export type CreateKernelFn = (
config: NerveConfig,
nerveRoot: string,
opts: { enableFileWatcher: boolean; ipcSocketPath: string },
) => {
groups: Set<string>;
ready: Promise<void>;
stop: () => Promise<void>;
};
function readConfig(nerveRoot: string): ReturnType<typeof parseNerveConfig> {
const configPath = join(nerveRoot, "nerve.yaml");
let raw: string;
try {
raw = readFileSync(configPath, "utf8");
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
return { ok: false, error: new Error(`❌ Cannot read ${configPath}: ${msg}`) };
}
return parseNerveConfig(raw);
}
export async function runForegroundKernelSession(
nerveRoot: string,
createKernel: CreateKernelFn,
): Promise<void> {
const configResult = readConfig(nerveRoot);
if (!configResult.ok) {
process.stderr.write(`${configResult.error.message}\n`);
process.exit(1);
}
const config = configResult.value;
const kernel = createKernel(config, nerveRoot, {
enableFileWatcher: true,
ipcSocketPath: getSocketPath(),
});
const senseNames = Object.keys(config.senses);
const groups = [...kernel.groups];
process.stdout.write(
`✅ Nerve starting — ${senseNames.length} sense(s), ${groups.length} group(s)\n`,
);
for (const group of groups) {
const groupSenses = Object.entries(config.senses)
.filter(([, sc]) => sc.group === group)
.map(([name]) => name);
process.stdout.write(` group "${group}": ${groupSenses.join(", ")}\n`);
}
process.stdout.write(" Press Ctrl+C to stop.\n");
let shuttingDown = false;
async function shutdown(): Promise<void> {
if (shuttingDown) return;
shuttingDown = true;
process.stdout.write("\n[nerve] Shutting down…\n");
await kernel.stop();
process.exit(0);
}
process.on("SIGINT", () => {
shutdown().catch((e: unknown) => {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`[nerve] Shutdown error: ${msg}\n`);
process.exit(1);
});
});
process.on("SIGTERM", () => {
shutdown().catch((e: unknown) => {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`[nerve] Shutdown error: ${msg}\n`);
process.exit(1);
});
});
await kernel.ready;
}
+50
View File
@@ -0,0 +1,50 @@
import { existsSync } from "node:fs";
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { pathToFileURL } from "node:url";
import type { NerveConfig } from "@uncaged/nerve-core";
import type { LogStore } from "./daemon-types.js";
export function getDaemonEntryPath(nerveRoot: string): string | undefined {
const pkgPath = join(nerveRoot, "node_modules", "@uncaged", "nerve-daemon", "package.json");
if (!existsSync(pkgPath)) return undefined;
try {
const pkg = JSON.parse(readFileSync(pkgPath, "utf8")) as { main?: string };
const main = pkg.main ?? "dist/index.js";
return join(nerveRoot, "node_modules", "@uncaged", "nerve-daemon", main);
} catch {
return join(nerveRoot, "node_modules", "@uncaged", "nerve-daemon", "dist", "index.js");
}
}
export function assertWorkspaceDaemonInstalled(nerveRoot: string): string {
const entry = getDaemonEntryPath(nerveRoot);
if (!entry || !existsSync(entry)) {
throw new Error(
`@uncaged/nerve-daemon is not installed under ${nerveRoot}/node_modules/. Run \`nerve init\` (or \`nerve init --force\`) to install workspace dependencies.`,
);
}
return entry;
}
/** Loaded from ~/.uncaged-nerve/node_modules at runtime — keep types structural only. */
export type DaemonModule = {
createKernel: (
config: NerveConfig,
nerveRoot: string,
options: { enableFileWatcher: boolean; ipcSocketPath: string },
) => {
groups: Set<string>;
ready: Promise<void>;
stop: () => Promise<void>;
};
createLogStore: (dbPath: string) => LogStore;
};
export async function loadDaemonModule(nerveRoot: string): Promise<DaemonModule> {
const entry = assertWorkspaceDaemonInstalled(nerveRoot);
const url = pathToFileURL(entry).href;
return import(url) as Promise<DaemonModule>;
}
+4
View File
@@ -10,6 +10,10 @@ export function getPidPath(): string {
return join(getNerveRoot(), "nerve.pid");
}
export function getSocketPath(): string {
return join(getNerveRoot(), "nerve.sock");
}
export function getLogPath(): string {
return join(getNerveRoot(), "logs", "nerve.log");
}
+6 -1
View File
@@ -1,8 +1,13 @@
import { defineConfig } from "tsup";
export default defineConfig({
entry: ["src/index.ts", "src/cli.ts"],
entry: ["src/index.ts", "src/cli.ts", "src/daemon-bootstrap.ts"],
format: ["esm"],
dts: true,
clean: true,
banner: {
js: "#!/usr/bin/env node",
},
/** Daemon is loaded from workspace node_modules at runtime — never bundle it. */
external: ["@uncaged/nerve-daemon"],
});
+2 -2
View File
@@ -1,11 +1,11 @@
{
"name": "@uncaged/nerve-core",
"version": "0.0.1",
"private": true,
"version": "0.1.4",
"type": "module",
"main": "dist/index.js",
"types": "dist/index.d.ts",
"scripts": {
"prepublishOnly": "bash ../../scripts/prepublish-check.sh",
"build": "tsup",
"test": "vitest run"
},
+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;
+8 -2
View File
@@ -1,11 +1,17 @@
{
"name": "@uncaged/nerve-daemon",
"version": "0.0.1",
"private": true,
"version": "0.1.5",
"type": "module",
"main": "dist/index.js",
"types": "dist/index.d.ts",
"files": [
"dist"
],
"publishConfig": {
"access": "public"
},
"scripts": {
"prepublishOnly": "bash ../../scripts/prepublish-check.sh",
"build": "tsup",
"test": "vitest run"
},
@@ -0,0 +1,105 @@
import { createHash } from "node:crypto";
import { existsSync, readdirSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { describe, expect, it } from "vitest";
import { createBlobStore, normalizeBlobHash } from "../blob-store.js";
function makeRoot(): string {
return join(tmpdir(), `nerve-blob-${Date.now()}-${Math.random().toString(16).slice(2)}`);
}
describe("normalizeBlobHash", () => {
it("accepts 64-char lowercase hex", () => {
const h = "a".repeat(64);
expect(normalizeBlobHash(h)).toBe(h);
});
it("normalizes uppercase to lowercase", () => {
const h = "A".repeat(64);
expect(normalizeBlobHash(h)).toBe("a".repeat(64));
});
it("rejects wrong length and non-hex", () => {
expect(normalizeBlobHash("ab")).toBeNull();
expect(normalizeBlobHash("g".repeat(64))).toBeNull();
});
});
describe("createBlobStore", () => {
it("write returns sha256 hex and stores under 2-char shard", () => {
const root = makeRoot();
const store = createBlobStore(root);
const content = "hello cas";
const hash = store.write(content);
expect(hash).toMatch(/^[0-9a-f]{64}$/);
expect(createHash("sha256").update(content, "utf8").digest("hex")).toBe(hash);
const shard = hash.slice(0, 2);
const rel = hash.slice(2);
const filePath = join(root, shard, rel);
expect(existsSync(filePath)).toBe(true);
});
it("read returns stored bytes and exists is true", () => {
const root = makeRoot();
const store = createBlobStore(root);
const buf = Buffer.from([0, 255, 128]);
const hash = store.write(buf);
expect(store.exists(hash)).toBe(true);
const got = store.read(hash);
expect(got).not.toBeNull();
expect(Buffer.compare(got as Buffer, buf)).toBe(0);
});
it("write is idempotent for same content", () => {
const root = makeRoot();
const store = createBlobStore(root);
const h1 = store.write("same");
const h2 = store.write("same");
expect(h1).toBe(h2);
const shard = h1.slice(0, 2);
const names = readdirSync(join(root, shard));
expect(names.filter((n: string) => !n.startsWith("."))).toHaveLength(1);
});
it("read returns null for missing blob", () => {
const root = makeRoot();
const store = createBlobStore(root);
const missing = "0".repeat(64);
expect(store.read(missing)).toBeNull();
expect(store.exists(missing)).toBe(false);
});
it("read and exists return null/false for invalid hash", () => {
const root = makeRoot();
const store = createBlobStore(root);
expect(store.read("not-a-hash")).toBeNull();
expect(store.exists("not-a-hash")).toBe(false);
});
it("throws when on-disk content does not match path hash", () => {
const root = makeRoot();
const store = createBlobStore(root);
const hash = store.write("ok");
const filePath = join(root, hash.slice(0, 2), hash.slice(2));
writeFileSync(filePath, "tampered");
expect(() => store.read(hash)).toThrow(/CAS mismatch/i);
});
it("write throws when an existing file at the digest path has wrong content", () => {
const root = makeRoot();
const store = createBlobStore(root);
const hash = store.write("truth");
const filePath = join(root, hash.slice(0, 2), hash.slice(2));
writeFileSync(filePath, "lies");
expect(() => store.write("truth")).toThrow(/CAS mismatch/i);
});
});
@@ -0,0 +1,430 @@
/**
* Phase 3 — Worker crash recovery tests.
*
* Verifies that WorkflowManager correctly:
* - Marks in-flight threads as "crashed" in the DB when a worker exits unexpectedly
* - Respawns the worker after a crash
* - Resumes "started" threads from persisted event history (resume-thread IPC)
* - Re-queues "queued" threads so they are dispatched on the new worker
*/
import { EventEmitter } from "node:events";
import type { NerveConfig, WorkflowConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
type MockChild = EventEmitter & {
send: ReturnType<typeof vi.fn>;
kill: ReturnType<typeof vi.fn>;
connected: boolean;
exitCode: number | null;
pid: number;
};
const mockChildren: MockChild[] = [];
function makeMockChild(pid = 1): MockChild {
const child = new EventEmitter() as MockChild;
child.connected = true;
child.exitCode = null;
child.pid = pid;
child.send = vi.fn((msg: unknown) => {
if (
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "shutdown"
) {
setImmediate(() => {
child.exitCode = 0;
child.connected = false;
child.emit("exit", 0, null);
});
}
});
child.kill = vi.fn((_signal?: string) => {
child.exitCode = 1;
child.connected = false;
child.emit("exit", null, _signal ?? "SIGKILL");
});
return child;
}
vi.mock("node:child_process", () => ({
fork: vi.fn((_script: string, _args: string[], _opts: unknown) => {
const child = makeMockChild(mockChildren.length + 1);
mockChildren.push(child);
return child;
}),
}));
const { createWorkflowManager } = await import("../workflow-manager.js");
function makeConfig(workflows: Record<string, WorkflowConfig> = {}): NerveConfig {
return {
senses: {},
reflexes: [],
workflows,
};
}
function makeLogStore(
activeRuns: Array<{
runId: string;
workflow: string;
status: "queued" | "started";
ts: number;
}> = [],
) {
const store = {
append: vi.fn(),
query: vi.fn(() => []),
getMeta: vi.fn(() => null),
setMeta: vi.fn(),
upsertWorkflowRun: vi.fn(),
appendWithWorkflowUpdate: vi.fn(),
getWorkflowRun: vi.fn(() => null),
getActiveWorkflowRuns: vi.fn((_workflowName?: string) => {
if (_workflowName !== undefined) {
return activeRuns.filter((r) => r.workflow === _workflowName);
}
return activeRuns;
}),
getTriggerPayload: vi.fn(() => ({ value: 42 })),
getThreadEvents: vi.fn(() => [{ type: "thread_start", triggerPayload: {} }]),
archiveLogs: vi.fn(() => ({ days: [], vacuumed: false })),
close: vi.fn(),
};
return store;
}
describe("WorkflowManager — crash recovery (Phase 3)", () => {
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
});
afterEach(() => {
vi.useRealTimers();
vi.clearAllMocks();
});
describe("worker crash marks active threads as crashed", () => {
it("logs 'crashed' status for each active thread when worker exits unexpectedly", async () => {
const logStore = makeLogStore();
const config = makeConfig({
"my-wf": { concurrency: 2, overflow: "drop" },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", { n: 1 });
mgr.startWorkflow("my-wf", { n: 2 });
expect(mgr.activeCount("my-wf")).toBe(2);
// Simulate unexpected exit (not shutdown)
const child = mockChildren[0];
child.exitCode = 1;
child.connected = false;
child.emit("exit", 1, null);
const crashedCalls = logStore.upsertWorkflowRun.mock.calls.filter(
([entry]: [{ type: string }]) => entry.type === "crashed",
);
expect(crashedCalls).toHaveLength(2);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("clears active count after crash", async () => {
const logStore = makeLogStore();
const config = makeConfig({
"my-wf": { concurrency: 3, overflow: "drop" },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", {});
mgr.startWorkflow("my-wf", {});
expect(mgr.activeCount("my-wf")).toBe(2);
const child = mockChildren[0];
child.exitCode = 1;
child.connected = false;
child.emit("exit", 1, null);
expect(mgr.activeCount("my-wf")).toBe(0);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("worker crash triggers respawn", () => {
it("spawns a new worker after crash", async () => {
const logStore = makeLogStore();
const config = makeConfig({
"my-wf": { concurrency: 1, overflow: "drop" },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", {});
expect(mockChildren).toHaveLength(1);
const child = mockChildren[0];
child.exitCode = 1;
child.connected = false;
child.emit("exit", 1, null);
// setImmediate to allow respawn
await vi.runAllTimersAsync();
expect(mockChildren).toHaveLength(2);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("sends resume-thread for 'started' runs from DB after respawn", async () => {
const activeRuns = [
{ runId: "run-started-1", workflow: "my-wf", status: "started" as const, ts: 1000 },
];
const logStore = makeLogStore(activeRuns);
logStore.getThreadEvents.mockReturnValue([
{ type: "thread_start", triggerPayload: {} },
{ type: "scan_complete", items: ["a"] },
]);
logStore.getTriggerPayload.mockReturnValue({ trigger: "initial" });
const config = makeConfig({
"my-wf": { concurrency: 2, overflow: "drop" },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", {});
const firstChild = mockChildren[0];
firstChild.exitCode = 1;
firstChild.connected = false;
firstChild.emit("exit", 1, null);
await vi.runAllTimersAsync();
// New worker should have been spawned
const secondChild = mockChildren[1];
expect(secondChild).toBeDefined();
// resume-thread should have been sent
const resumeCalls = (secondChild.send as ReturnType<typeof vi.fn>).mock.calls.filter(
([msg]: [unknown]) =>
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "resume-thread",
);
expect(resumeCalls).toHaveLength(1);
expect(resumeCalls[0][0]).toMatchObject({
type: "resume-thread",
runId: "run-started-1",
triggerPayload: { trigger: "initial" },
});
expect(Array.isArray((resumeCalls[0][0] as Record<string, unknown>).events)).toBe(true);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("re-queues 'queued' runs from DB after respawn", async () => {
const activeRuns = [
{ runId: "run-queued-1", workflow: "my-wf", status: "queued" as const, ts: 900 },
];
const logStore = makeLogStore(activeRuns);
logStore.getTriggerPayload.mockReturnValue({ queued: "payload" });
const config = makeConfig({
"my-wf": { concurrency: 1, overflow: "queue", maxQueue: 5 },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
// Start one thread to fill the concurrency slot (so queued run stays queued on respawn)
mgr.startWorkflow("my-wf", {});
const firstChild = mockChildren[0];
firstChild.exitCode = 1;
firstChild.connected = false;
firstChild.emit("exit", 1, null);
await vi.runAllTimersAsync();
// After respawn, the queue should contain the recovered run
expect(mgr.queueLength("my-wf")).toBeGreaterThanOrEqual(1);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("command events are persisted (for crash recovery replay)", () => {
it("persists thread_command_event when worker sends thread-command-event IPC", async () => {
const logStore = makeLogStore();
const config = makeConfig({
"my-wf": { concurrency: 1, overflow: "drop" },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", { x: 1 });
const child = mockChildren[0];
const startCall = (child.send as ReturnType<typeof vi.fn>).mock.calls[0];
const runId = (startCall[0] as Record<string, unknown>).runId as string;
// Simulate worker sending a command event back
child.emit("message", {
type: "thread-command-event",
runId,
event: { type: "scan_complete", items: ["a", "b"] },
});
const appendCalls = logStore.append.mock.calls.filter(
([entry]: [{ type: string }]) => entry.type === "thread_command_event",
);
expect(appendCalls).toHaveLength(1);
expect(appendCalls[0][0]).toMatchObject({
source: "workflow",
type: "thread_command_event",
refId: runId,
});
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("triggerPayload is persisted in 'started' log entry", () => {
it("stores triggerPayload in the payload field of the started log entry", async () => {
const logStore = makeLogStore();
const config = makeConfig({
"my-wf": { concurrency: 1, overflow: "drop" },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
const payload = { task: "build-docker", repo: "myrepo" };
mgr.startWorkflow("my-wf", payload);
const startedCall = logStore.upsertWorkflowRun.mock.calls.find(
([entry]: [{ type: string }]) => entry.type === "started",
);
expect(startedCall).toBeDefined();
const logEntry = startedCall?.[0] as { payload: string | null };
expect(logEntry.payload).not.toBeNull();
const parsed = JSON.parse(logEntry.payload as string) as Record<string, unknown>;
expect(parsed.triggerPayload).toMatchObject(payload);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("runId deduplication in crash recovery", () => {
it("does not push duplicate runIds into the queue during crash recovery", async () => {
const activeRuns = [
{ runId: "run-queued-dup", workflow: "my-wf", status: "queued" as const, ts: 900 },
];
const logStore = makeLogStore(activeRuns);
logStore.getTriggerPayload.mockReturnValue({ q: 1 });
const config = makeConfig({
"my-wf": { concurrency: 1, overflow: "queue", maxQueue: 5 },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
// Start one thread to fill the concurrency slot
mgr.startWorkflow("my-wf", {});
const firstChild = mockChildren[0];
// Crash once → respawn → crash again → second respawn
firstChild.exitCode = 1;
firstChild.connected = false;
firstChild.emit("exit", 1, null);
await vi.runAllTimersAsync();
const secondChild = mockChildren[1];
secondChild.exitCode = 1;
secondChild.connected = false;
secondChild.emit("exit", 1, null);
await vi.runAllTimersAsync();
// The recovered queued run should appear at most once in the queue
expect(mgr.queueLength("my-wf")).toBeLessThanOrEqual(1);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("does not add duplicate active runIds during crash recovery", async () => {
const activeRuns = [
{ runId: "run-started-dup", workflow: "my-wf", status: "started" as const, ts: 1000 },
];
const logStore = makeLogStore(activeRuns);
logStore.getThreadEvents.mockReturnValue([{ type: "thread_start", triggerPayload: {} }]);
logStore.getTriggerPayload.mockReturnValue({ s: 1 });
const config = makeConfig({
"my-wf": { concurrency: 2, overflow: "drop" },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", {});
const firstChild = mockChildren[0];
firstChild.exitCode = 1;
firstChild.connected = false;
firstChild.emit("exit", 1, null);
await vi.runAllTimersAsync();
const secondChild = mockChildren[1];
secondChild.exitCode = 1;
secondChild.connected = false;
secondChild.emit("exit", 1, null);
await vi.runAllTimersAsync();
// The active set should not double-count the recovered run
expect(mgr.activeCount("my-wf")).toBeLessThanOrEqual(1);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("crash-loop backoff", () => {
it("stops respawning after exceeding max crashes in the window", async () => {
const logStore = makeLogStore();
const config = makeConfig({
"crash-wf": { concurrency: 1, overflow: "drop" },
});
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("crash-wf", {});
// Crash the worker 6 times in rapid succession (within CRASH_WINDOW_MS = 60s)
for (let i = 0; i < 6; i++) {
const child = mockChildren[mockChildren.length - 1];
child.exitCode = 1;
child.connected = false;
child.emit("exit", 1, null);
await vi.runAllTimersAsync();
}
// After 6 crashes, no new worker should be spawned
// The 1st crash spawns child[1], ..., 5th crash spawns child[5], 6th should NOT spawn
expect(mockChildren.length).toBeLessThanOrEqual(6);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
});
@@ -0,0 +1,234 @@
/**
* Unit + integration tests for daemon-ipc.ts — trigger-sense request type.
*
* Tests cover:
* - parseRequest correctly accepts/rejects trigger-sense messages
* - createDaemonIpcServer routes trigger-sense to opts.triggerSense
* - Error response when triggerSense throws (unknown sense)
* - Success response on valid sense trigger
*/
import { rmSync } from "node:fs";
import { connect } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { createDaemonIpcServer } from "../daemon-ipc.js";
import type { DaemonIpcServer } from "../daemon-ipc.js";
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
let sockPath: string;
let server: DaemonIpcServer | null = null;
function makeMockWorkflowManager() {
return {
startWorkflow: vi.fn(),
stop: vi.fn(async () => {}),
totalActiveCount: vi.fn(() => 0),
drainAndRespawn: vi.fn(async () => {}),
updateConfig: vi.fn(),
getActiveWorkflowRuns: vi.fn(() => []),
};
}
function sendRaw(path: string, message: object): Promise<object> {
return new Promise((resolve, reject) => {
const sock = connect(path, () => {
let buf = "";
sock.on("data", (chunk: Buffer) => {
buf += chunk.toString("utf8");
const lines = buf.split("\n");
for (const line of lines) {
const trimmed = line.trim();
if (trimmed.length === 0) continue;
try {
resolve(JSON.parse(trimmed) as object);
} catch {
reject(new Error(`Invalid JSON response: ${trimmed}`));
}
sock.destroy();
return;
}
buf = lines[lines.length - 1] ?? "";
});
sock.write(`${JSON.stringify(message)}\n`);
});
sock.on("error", reject);
});
}
beforeEach(() => {
sockPath = join(tmpdir(), `nerve-ipc-test-${Date.now()}-${Math.random().toString(36).slice(2)}.sock`);
});
afterEach(async () => {
if (server !== null) {
await server.close();
server = null;
}
try {
rmSync(sockPath);
} catch {
// already removed
}
});
// ---------------------------------------------------------------------------
// trigger-sense: valid request → ok: true
// ---------------------------------------------------------------------------
describe("daemon-ipc — trigger-sense", () => {
it("responds ok:true when triggerSense succeeds", async () => {
const triggerSense = vi.fn();
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense,
listSenses: vi.fn(() => []),
});
const resp = await sendRaw(sockPath, { type: "trigger-sense", sense: "cpu-usage" });
expect(resp).toEqual({ ok: true });
expect(triggerSense).toHaveBeenCalledOnce();
expect(triggerSense).toHaveBeenCalledWith("cpu-usage");
});
it("responds ok:false with error message when triggerSense throws", async () => {
const triggerSense = vi.fn(() => {
throw new Error('Unknown sense: "no-such-sense"');
});
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense,
listSenses: vi.fn(() => []),
});
const resp = await sendRaw(sockPath, { type: "trigger-sense", sense: "no-such-sense" });
expect(resp).toEqual({ ok: false, error: 'Unknown sense: "no-such-sense"' });
expect(triggerSense).toHaveBeenCalledWith("no-such-sense");
});
it("responds ok:false for trigger-sense with empty sense name", async () => {
const triggerSense = vi.fn();
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense,
listSenses: vi.fn(() => []),
});
const resp = await sendRaw(sockPath, { type: "trigger-sense", sense: "" });
expect(resp).toEqual({ ok: false, error: "Invalid request" });
expect(triggerSense).not.toHaveBeenCalled();
});
it("responds ok:false for trigger-sense missing sense field", async () => {
const triggerSense = vi.fn();
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense,
listSenses: vi.fn(() => []),
});
const resp = await sendRaw(sockPath, { type: "trigger-sense" });
expect(resp).toEqual({ ok: false, error: "Invalid request" });
expect(triggerSense).not.toHaveBeenCalled();
});
it("does NOT call triggerSense for trigger-workflow requests", async () => {
const triggerSense = vi.fn();
const wfManager = makeMockWorkflowManager();
server = createDaemonIpcServer(sockPath, wfManager as never, {
triggerSense,
listSenses: vi.fn(() => []),
});
const resp = await sendRaw(sockPath, {
type: "trigger-workflow",
workflow: "my-workflow",
payload: {},
});
expect(resp).toEqual({ ok: true });
expect(triggerSense).not.toHaveBeenCalled();
expect(wfManager.startWorkflow).toHaveBeenCalledWith("my-workflow", {});
});
it("responds ok:false for completely unknown request type", async () => {
const triggerSense = vi.fn();
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense,
listSenses: vi.fn(() => []),
});
const resp = await sendRaw(sockPath, { type: "unknown-type", data: "x" });
expect(resp).toEqual({ ok: false, error: "Invalid request" });
expect(triggerSense).not.toHaveBeenCalled();
});
});
// ---------------------------------------------------------------------------
// list-senses: valid request → ok: true with senses array
// ---------------------------------------------------------------------------
describe("daemon-ipc — list-senses", () => {
it("responds ok:true with empty senses array when listSenses returns []", async () => {
const listSenses = vi.fn(() => []);
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense: vi.fn(),
listSenses,
});
const resp = await sendRaw(sockPath, { type: "list-senses" });
expect(resp).toEqual({ ok: true, senses: [] });
expect(listSenses).toHaveBeenCalledOnce();
});
it("responds ok:true with senses populated from listSenses", async () => {
const sensesData = [
{ name: "cpu-usage", group: "system", throttle: 5000, timeout: 3000, lastSignalTs: 1000 },
{ name: "disk-usage", group: "system", throttle: 30000, timeout: null, lastSignalTs: null },
];
const listSenses = vi.fn(() => sensesData);
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense: vi.fn(),
listSenses,
});
const resp = await sendRaw(sockPath, { type: "list-senses" });
expect(resp).toEqual({ ok: true, senses: sensesData });
expect(listSenses).toHaveBeenCalledOnce();
});
it("responds ok:false when listSenses throws", async () => {
const listSenses = vi.fn(() => {
throw new Error("internal error");
});
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense: vi.fn(),
listSenses,
});
const resp = await sendRaw(sockPath, { type: "list-senses" });
expect(resp).toEqual({ ok: false, error: "internal error" });
});
it("does NOT call listSenses for trigger-sense requests", async () => {
const listSenses = vi.fn(() => []);
server = createDaemonIpcServer(sockPath, makeMockWorkflowManager() as never, {
triggerSense: vi.fn(),
listSenses,
});
await sendRaw(sockPath, { type: "trigger-sense", sense: "cpu-usage" });
expect(listSenses).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,119 @@
/**
* Phase 3 — FileWatcher workflow change detection tests.
*
* Verifies that file-watcher.ts detects .ts file changes under workflows/.
*/
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, describe, expect, it } from "vitest";
import { createFileWatcher } from "../file-watcher.js";
import type { FileChange, FileWatcher } from "../file-watcher.js";
function makeTempNerveRoot(): string {
const dir = mkdtempSync(join(tmpdir(), "nerve-fw-wf-test-"));
mkdirSync(join(dir, "workflows", "my-workflow"), { recursive: true });
writeFileSync(join(dir, "nerve.yaml"), "senses: {}\nreflexes: []\n");
writeFileSync(
join(dir, "workflows", "my-workflow", "index.ts"),
"export default { roles: {}, moderate: () => null };",
);
return dir;
}
async function waitFor(
predicate: () => boolean,
timeoutMs: number,
intervalMs = 50,
): Promise<void> {
return new Promise((resolve, reject) => {
const timer = setTimeout(
() => reject(new Error(`waitFor timed out after ${timeoutMs}ms`)),
timeoutMs,
);
const check = setInterval(() => {
if (predicate()) {
clearTimeout(timer);
clearInterval(check);
resolve();
}
}, intervalMs);
});
}
describe("createFileWatcher — workflow file changes (Phase 3)", () => {
let watcher: FileWatcher | null = null;
afterEach(() => {
if (watcher !== null) {
watcher.close();
watcher = null;
}
});
it("detects workflow .ts file changes and emits kind=workflow", async () => {
const root = makeTempNerveRoot();
const changes: FileChange[] = [];
watcher = createFileWatcher(root, (change) => changes.push(change), 50);
await new Promise((r) => setTimeout(r, 100));
writeFileSync(
join(root, "workflows", "my-workflow", "index.ts"),
"export default { roles: {}, moderate: () => null }; // updated",
);
await waitFor(() => changes.some((c) => c.kind === "workflow"), 3000);
const wfChanges = changes.filter((c) => c.kind === "workflow");
expect(wfChanges.length).toBeGreaterThanOrEqual(1);
const wfChange = wfChanges[0] as { workflowName: string; filePath: string };
expect(wfChange.workflowName).toBe("my-workflow");
}, 10_000);
it("does NOT emit workflow change for nerve.yaml", async () => {
const root = makeTempNerveRoot();
const changes: FileChange[] = [];
watcher = createFileWatcher(root, (change) => changes.push(change), 50);
await new Promise((r) => setTimeout(r, 100));
writeFileSync(join(root, "nerve.yaml"), "senses: {}\nreflexes: []\n# changed\n");
await waitFor(() => changes.some((c) => c.kind === "config"), 3000);
const wfChanges = changes.filter((c) => c.kind === "workflow");
expect(wfChanges).toHaveLength(0);
}, 10_000);
it("debounces rapid workflow file changes", async () => {
const root = makeTempNerveRoot();
const changes: FileChange[] = [];
watcher = createFileWatcher(root, (change) => changes.push(change), 200);
await new Promise((r) => setTimeout(r, 100));
for (let i = 0; i < 5; i++) {
writeFileSync(
join(root, "workflows", "my-workflow", "index.ts"),
`export default {}; // v${i}`,
);
}
await waitFor(() => changes.some((c) => c.kind === "workflow"), 3000);
// Allow debounce window to pass
await new Promise((r) => setTimeout(r, 300));
const wfChanges = changes.filter((c) => c.kind === "workflow");
expect(wfChanges.length).toBe(1);
}, 10_000);
it("cleans up temp dir after test", () => {
const root = makeTempNerveRoot();
rmSync(root, { recursive: true, force: true });
});
});
@@ -0,0 +1,354 @@
/**
* Phase 3 — Hot reload tests.
*
* Verifies that:
* - drainAndRespawn() sends shutdown, waits for exit, then respawns the worker
* - Kernel dispatches handleWorkflowFileChange when file-watcher emits a workflow change
* - Kernel logs a workflow_reload system event on hot reload
* - drainAndRespawn on a non-existent worker is a no-op
* - drainAndRespawn after the drain sends a fresh worker (not crash-recovery)
*/
import { EventEmitter } from "node:events";
import type { NerveConfig, WorkflowConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
type MockChild = EventEmitter & {
send: ReturnType<typeof vi.fn>;
kill: ReturnType<typeof vi.fn>;
connected: boolean;
exitCode: number | null;
pid: number;
};
const mockChildren: MockChild[] = [];
function makeMockChild(pid = 1): MockChild {
const child = new EventEmitter() as MockChild;
child.connected = true;
child.exitCode = null;
child.pid = pid;
child.send = vi.fn((msg: unknown) => {
if (
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "shutdown"
) {
setImmediate(() => {
child.exitCode = 0;
child.connected = false;
child.emit("exit", 0, null);
});
}
});
child.kill = vi.fn((_signal?: string) => {
child.exitCode = 1;
child.connected = false;
child.emit("exit", null, _signal ?? "SIGKILL");
});
return child;
}
vi.mock("node:child_process", () => ({
fork: vi.fn((_script: string, _args: string[], _opts: unknown) => {
const child = makeMockChild(mockChildren.length + 1);
mockChildren.push(child);
return child;
}),
}));
const { createWorkflowManager } = await import("../workflow-manager.js");
const { createKernel } = await import("../kernel.js");
function makeWfConfig(workflows: Record<string, WorkflowConfig> = {}): NerveConfig {
return { senses: {}, reflexes: [], workflows };
}
function makeLogStore() {
return {
append: vi.fn(),
query: vi.fn(() => []),
getMeta: vi.fn(() => null),
setMeta: vi.fn(),
upsertWorkflowRun: vi.fn(),
appendWithWorkflowUpdate: vi.fn(),
getWorkflowRun: vi.fn(() => null),
getActiveWorkflowRuns: vi.fn(() => []),
getTriggerPayload: vi.fn(() => null),
getThreadEvents: vi.fn(() => []),
archiveLogs: vi.fn(() => ({ days: [], vacuumed: false })),
close: vi.fn(),
};
}
describe("WorkflowManager — drainAndRespawn (Phase 3 hot reload)", () => {
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
});
afterEach(() => {
vi.useRealTimers();
vi.clearAllMocks();
});
it("drainAndRespawn does NOT respawn when workflow is removed from config", async () => {
const logStore = makeLogStore();
const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", {});
expect(mockChildren).toHaveLength(1);
// Remove workflow from config before drain completes
mgr.updateConfig(makeWfConfig({}));
const drainPromise = mgr.drainAndRespawn("my-wf", 5000);
await vi.runAllTimersAsync();
await drainPromise;
// No new worker should have been spawned (workflow was removed)
expect(mockChildren).toHaveLength(1);
});
it("drainAndRespawn marks in-flight runs as interrupted in DB", async () => {
const logStore = makeLogStore();
const config = makeWfConfig({ "my-wf": { concurrency: 2, overflow: "drop" } });
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", { n: 1 });
mgr.startWorkflow("my-wf", { n: 2 });
expect(mgr.activeCount("my-wf")).toBe(2);
const drainPromise = mgr.drainAndRespawn("my-wf", 5000);
await vi.runAllTimersAsync();
await drainPromise;
const interruptedCalls = logStore.upsertWorkflowRun.mock.calls.filter(
([entry]: [{ type: string }]) => entry.type === "interrupted",
);
expect(interruptedCalls).toHaveLength(2);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("drainAndRespawn on an unknown workflow (no worker) resolves immediately", async () => {
const logStore = makeLogStore();
const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
const mgr = createWorkflowManager("/nerve-root", config, logStore);
// No thread started — no worker spawned
await expect(mgr.drainAndRespawn("my-wf")).resolves.toBeUndefined();
});
it("drainAndRespawn sends shutdown to existing worker and waits for exit", async () => {
const logStore = makeLogStore();
const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", {});
expect(mockChildren).toHaveLength(1);
const drainPromise = mgr.drainAndRespawn("my-wf", 5000);
await vi.runAllTimersAsync();
await drainPromise;
const firstChild = mockChildren[0];
expect(firstChild.send).toHaveBeenCalledWith(expect.objectContaining({ type: "shutdown" }));
});
it("drainAndRespawn spawns a fresh worker after the old one exits", async () => {
const logStore = makeLogStore();
const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", {});
expect(mockChildren).toHaveLength(1);
const drainPromise = mgr.drainAndRespawn("my-wf", 5000);
await vi.runAllTimersAsync();
await drainPromise;
// A new worker should have been spawned (not crash-recovery, just fresh)
expect(mockChildren).toHaveLength(2);
});
it("fresh worker after drainAndRespawn does NOT receive resume-thread messages", async () => {
const logStore = makeLogStore();
// Even if there are active runs in DB, after drain the worker should NOT get resume
const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", {});
const drainPromise = mgr.drainAndRespawn("my-wf", 5000);
await vi.runAllTimersAsync();
await drainPromise;
const newChild = mockChildren[1];
const resumeCalls = (newChild.send as ReturnType<typeof vi.fn>).mock.calls.filter(
([msg]: [unknown]) =>
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "resume-thread",
);
expect(resumeCalls).toHaveLength(0);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("new threads can be started on the fresh worker after drainAndRespawn", async () => {
const logStore = makeLogStore();
const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
const mgr = createWorkflowManager("/nerve-root", config, logStore);
mgr.startWorkflow("my-wf", { first: true });
const drainPromise = mgr.drainAndRespawn("my-wf", 5000);
await vi.runAllTimersAsync();
await drainPromise;
// Start a new thread on the fresh worker
mgr.startWorkflow("my-wf", { second: true });
const newChild = mockChildren[1];
const startCalls = (newChild.send as ReturnType<typeof vi.fn>).mock.calls.filter(
([msg]: [unknown]) =>
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "start-thread",
);
expect(startCalls).toHaveLength(1);
const stopPromise = mgr.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("Kernel — workflow hot reload via file-watcher (Phase 3)", () => {
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
});
afterEach(async () => {
vi.useRealTimers();
vi.clearAllMocks();
});
it("handleWorkflowFileChange logs workflow_reload system event", async () => {
const logStore = makeLogStore();
const config: NerveConfig = {
senses: {},
reflexes: [{ kind: "workflow", workflow: "my-wf", on: null }],
workflows: { "my-wf": { concurrency: 1, overflow: "drop" } },
};
const kernel = createKernel(config, "/tmp/nerve-hot-reload-test", {
workerScript: "fake-worker.js",
logStore,
});
// Trigger a workflow thread so a worker is spawned
kernel.workflowManager.startWorkflow("my-wf", {});
// Manually call drainAndRespawn (simulating what kernel does on workflow file change)
const drainPromise = kernel.workflowManager.drainAndRespawn("my-wf", 1000);
await vi.runAllTimersAsync();
await drainPromise;
// Kernel's handleWorkflowFileChange should log a workflow_reload event
// We test this via the kernel itself
const appendCalls = logStore.append.mock.calls;
const startCall = appendCalls.find(([e]: [{ type: string }]) => e.type === "start");
expect(startCall).toBeDefined();
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("reloadConfig drains worker for removed workflows", async () => {
const logStore = makeLogStore();
const initialConfig: NerveConfig = {
senses: {},
reflexes: [{ kind: "workflow", workflow: "old-wf", on: null }],
workflows: { "old-wf": { concurrency: 1, overflow: "drop" } },
};
const kernel = createKernel(initialConfig, "/tmp/nerve-hot-reload-test", {
workerScript: "fake-worker.js",
logStore,
});
// Spawn a worker for old-wf
kernel.workflowManager.startWorkflow("old-wf", {});
expect(mockChildren).toHaveLength(1);
// Reload config without old-wf
const newConfig: NerveConfig = {
senses: {},
reflexes: [],
workflows: null,
};
kernel.reloadConfig(newConfig);
await vi.runAllTimersAsync();
// The old worker should have received a shutdown (drain)
expect(mockChildren[0].send).toHaveBeenCalledWith(
expect.objectContaining({ type: "shutdown" }),
);
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("reloadConfig updates concurrency/overflow without restarting worker", async () => {
const logStore = makeLogStore();
const initialConfig: NerveConfig = {
senses: {},
reflexes: [{ kind: "workflow", workflow: "my-wf", on: null }],
workflows: { "my-wf": { concurrency: 1, overflow: "drop" } },
};
const kernel = createKernel(initialConfig, "/tmp/nerve-hot-reload-test", {
workerScript: "fake-worker.js",
logStore,
});
kernel.workflowManager.startWorkflow("my-wf", {});
const workersBefore = mockChildren.length;
// Reload with updated concurrency — should NOT spawn a new workflow worker
const newConfig: NerveConfig = {
senses: {},
reflexes: [{ kind: "workflow", workflow: "my-wf", on: null }],
workflows: { "my-wf": { concurrency: 5, overflow: "queue", maxQueue: 50 } },
};
kernel.reloadConfig(newConfig);
// No extra workflow worker spawn (the config update is in-place)
// The worker count may increase if senses change, but the workflow worker should not be respawned
expect(mockChildren).toHaveLength(workersBefore);
// After reload, the new concurrency should be respected
expect(kernel.workflowManager.activeCount("my-wf")).toBe(1);
// Can now start up to 5 concurrent threads (previously only 1)
kernel.workflowManager.startWorkflow("my-wf", { n: 2 });
kernel.workflowManager.startWorkflow("my-wf", { n: 3 });
expect(kernel.workflowManager.activeCount("my-wf")).toBe(3);
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
@@ -0,0 +1,200 @@
/**
* Unit tests for kernel.triggerSense() — IPC issue #36.
*
* These tests use a mock child_process and a mock LogStore so they do NOT
* require better-sqlite3 to be present in the test environment.
*/
import { EventEmitter } from "node:events";
import type { NerveConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
// ---------------------------------------------------------------------------
// Mock child_process.fork before importing kernel
// ---------------------------------------------------------------------------
const mockChildren: MockChild[] = [];
type MockChild = EventEmitter & {
send: ReturnType<typeof vi.fn>;
kill: ReturnType<typeof vi.fn>;
connected: boolean;
pid: number;
};
function makeMockChild(pid = 1): MockChild {
const child = new EventEmitter() as MockChild;
child.connected = true;
child.send = vi.fn((msg: unknown) => {
if (
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "shutdown"
) {
setImmediate(() => {
child.connected = false;
child.emit("exit", 0, null);
});
}
});
child.kill = vi.fn((_signal?: string) => {
child.connected = false;
child.emit("exit", null, _signal ?? "SIGKILL");
});
child.pid = pid;
return child;
}
vi.mock("node:child_process", () => ({
fork: vi.fn((_script: string, _args: string[], _opts: unknown) => {
const child = makeMockChild(mockChildren.length + 1);
mockChildren.push(child);
return child;
}),
}));
// Import after mock is set up
const { createKernel } = await import("../kernel.js");
// ---------------------------------------------------------------------------
// Mock LogStore factory (avoids better-sqlite3 dependency)
// ---------------------------------------------------------------------------
function makeMockLogStore() {
return {
append: vi.fn(),
query: vi.fn(() => []),
getMeta: vi.fn(() => null),
setMeta: vi.fn(),
upsertWorkflowRun: vi.fn(),
appendWithWorkflowUpdate: vi.fn(),
getWorkflowRun: vi.fn(() => null),
getActiveWorkflowRuns: vi.fn(() => []),
getAllWorkflowRuns: vi.fn(() => []),
getTriggerPayload: vi.fn(() => null),
getThreadEvents: vi.fn(() => []),
archiveLogs: vi.fn(() => ({ days: [], vacuumed: false })),
close: vi.fn(),
};
}
// ---------------------------------------------------------------------------
// Config helpers
// ---------------------------------------------------------------------------
function makeConfig(overrides: Partial<NerveConfig> = {}): NerveConfig {
return {
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [],
workflows: null,
...overrides,
};
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
describe("kernel.triggerSense()", () => {
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
});
afterEach(() => {
vi.useRealTimers();
});
it("throws for an unknown sense name", async () => {
const config = makeConfig();
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: null,
logStore: makeMockLogStore() as never,
});
expect(() => kernel.triggerSense("no-such-sense")).toThrow(/Unknown sense/);
await kernel.stop();
});
it("sends a compute message to the worker for the correct group", async () => {
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
"net-io": { group: "network", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [],
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: null,
logStore: makeMockLogStore() as never,
});
// Two groups → two workers
expect(mockChildren.length).toBe(2);
// Workers are keyed by group: groups iteration order matches the insertion
// order from Object.values(config.senses). Find the worker for "system".
const systemWorkerIdx = Array.from(kernel.groups).indexOf("system");
const systemWorker = mockChildren[systemWorkerIdx];
kernel.triggerSense("cpu-usage");
expect(systemWorker.send).toHaveBeenCalledWith(
expect.objectContaining({ type: "compute", sense: "cpu-usage" }),
);
await kernel.stop();
});
it("sends a compute message to the correct worker when multiple senses share a group", async () => {
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
"disk-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [],
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: null,
logStore: makeMockLogStore() as never,
});
// Both senses share the "system" group → one worker only
expect(mockChildren.length).toBe(1);
const worker = mockChildren[0];
kernel.triggerSense("disk-usage");
expect(worker.send).toHaveBeenCalledWith(
expect.objectContaining({ type: "compute", sense: "disk-usage" }),
);
await kernel.stop();
});
it("does not send to a disconnected worker (does not throw)", async () => {
// Use real timers so kernel.stop() waitForExit can rely on SIGKILL timeout
vi.useRealTimers();
const config = makeConfig();
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: null,
logStore: makeMockLogStore() as never,
});
const worker = mockChildren[0];
worker.connected = false;
// Should not throw even when the worker is disconnected
expect(() => kernel.triggerSense("cpu-usage")).not.toThrow();
expect(worker.send).not.toHaveBeenCalledWith(
expect.objectContaining({ type: "compute" }),
);
await kernel.stop();
}, 10_000);
});
@@ -0,0 +1,421 @@
/**
* Integration tests for Kernel + WorkflowManager integration.
*
* Verifies that sense signals trigger workflow runs via workflow reflexes,
* that workflow events are logged, that reloadConfig handles workflow changes,
* and that graceful shutdown stops workflow workers.
*
* Uses mocked child_process.fork to avoid real subprocesses.
*/
import { EventEmitter } from "node:events";
import type { NerveConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
// ---------------------------------------------------------------------------
// Mock child_process.fork before importing kernel
// ---------------------------------------------------------------------------
type MockChild = EventEmitter & {
send: ReturnType<typeof vi.fn>;
kill: ReturnType<typeof vi.fn>;
connected: boolean;
exitCode: number | null;
pid: number;
};
const mockChildren: MockChild[] = [];
function makeMockChild(pid = 1): MockChild {
const child = new EventEmitter() as MockChild;
child.connected = true;
child.exitCode = null;
child.pid = pid;
child.send = vi.fn((msg: unknown) => {
if (
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "shutdown"
) {
setImmediate(() => {
child.exitCode = 0;
child.connected = false;
child.emit("exit", 0, null);
});
}
});
child.kill = vi.fn((_signal?: string) => {
child.exitCode = 1;
child.connected = false;
child.emit("exit", null, _signal ?? "SIGKILL");
});
return child;
}
vi.mock("node:child_process", () => ({
fork: vi.fn((_script: string, _args: string[], _opts: unknown) => {
const child = makeMockChild(mockChildren.length + 1);
mockChildren.push(child);
return child;
}),
}));
// Import after mock is set up
const { createKernel } = await import("../kernel.js");
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
function makeLogStore() {
return {
append: vi.fn(),
query: vi.fn(() => []),
getMeta: vi.fn(() => null),
setMeta: vi.fn(),
upsertWorkflowRun: vi.fn(),
appendWithWorkflowUpdate: vi.fn(),
getWorkflowRun: vi.fn(() => null),
getActiveWorkflowRuns: vi.fn(() => []),
getTriggerPayload: vi.fn(() => null),
getThreadEvents: vi.fn(() => []),
archiveLogs: vi.fn(() => ({ days: [], vacuumed: false })),
close: vi.fn(),
};
}
function makeConfig(overrides: Partial<NerveConfig> = {}): NerveConfig {
return {
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [],
workflows: null,
...overrides,
};
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
describe("kernel + workflowManager integration", () => {
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
});
afterEach(async () => {
vi.useRealTimers();
vi.clearAllMocks();
});
describe("sense signal triggers workflow via reflex", () => {
it("calls workflowManager.startWorkflow when a sense signal fires on a workflow reflex", async () => {
const logStore = makeLogStore();
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [{ kind: "workflow", workflow: "my-workflow", on: ["cpu-usage"] }],
workflows: { "my-workflow": { concurrency: 2, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
// Emit a signal from "cpu-usage" on the bus
const { createSignalBus } = await import("../signal-bus.js");
void createSignalBus; // ensure import resolves
kernel.bus.emit({ id: 1, senseId: "cpu-usage", payload: { value: 80 }, ts: Date.now() });
// The workflow worker should be spawned (one for the sense group, one for workflow)
// The sense group worker is mockChildren[0]; the workflow worker is mockChildren[1]
// We need to check that a start-thread message was sent to the workflow worker
const workflowWorker = mockChildren.find((c) =>
(c.send as ReturnType<typeof vi.fn>).mock.calls.some(
([msg]: [unknown]) =>
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "start-thread",
),
);
expect(workflowWorker).toBeDefined();
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("passes the signal payload as triggerPayload to the workflow", async () => {
const logStore = makeLogStore();
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [{ kind: "workflow", workflow: "alert-workflow", on: ["cpu-usage"] }],
workflows: { "alert-workflow": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
const payload = { level: "critical", value: 99 };
kernel.bus.emit({ id: 1, senseId: "cpu-usage", payload, ts: Date.now() });
// Find the start-thread call and verify triggerPayload
const startThreadCall = mockChildren
.flatMap((c) => (c.send as ReturnType<typeof vi.fn>).mock.calls as [unknown][])
.find(
([msg]) =>
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "start-thread",
);
expect(startThreadCall).toBeDefined();
expect(startThreadCall?.[0]).toMatchObject({
type: "start-thread",
workflow: "alert-workflow",
triggerPayload: payload,
});
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("does not trigger workflow when signal senseId is not in 'on' list", async () => {
const logStore = makeLogStore();
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
"disk-io": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [{ kind: "workflow", workflow: "my-workflow", on: ["disk-io"] }],
workflows: { "my-workflow": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
// Emit signal from cpu-usage — NOT in the workflow's "on" list
kernel.bus.emit({ id: 1, senseId: "cpu-usage", payload: 50, ts: Date.now() });
// No workflow worker should have been spawned (only the sense group worker)
const workflowWorkerSpawned = mockChildren.some((c) =>
(c.send as ReturnType<typeof vi.fn>).mock.calls.some(
([msg]: [unknown]) =>
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "start-thread",
),
);
expect(workflowWorkerSpawned).toBe(false);
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("workflow events are logged", () => {
it("logs a 'started' event when workflow thread is triggered", async () => {
const logStore = makeLogStore();
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [{ kind: "workflow", workflow: "log-test-workflow", on: ["cpu-usage"] }],
workflows: { "log-test-workflow": { concurrency: 2, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
kernel.bus.emit({ id: 1, senseId: "cpu-usage", payload: null, ts: Date.now() });
expect(logStore.upsertWorkflowRun).toHaveBeenCalledWith(
expect.objectContaining({ source: "workflow", type: "started" }),
expect.objectContaining({ workflow: "log-test-workflow", status: "started" }),
);
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("reloadConfig handles workflow changes", () => {
it("new workflow reflexes are active after reloadConfig", async () => {
const logStore = makeLogStore();
const initialConfig = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [],
workflows: null,
});
const kernel = createKernel(initialConfig, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
// Reload with a workflow reflex added
const newConfig: NerveConfig = {
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [{ kind: "workflow", workflow: "new-workflow", on: ["cpu-usage"] }],
workflows: { "new-workflow": { concurrency: 1, overflow: "drop" } },
};
kernel.reloadConfig(newConfig);
// Now emit a signal — should trigger the new workflow
kernel.bus.emit({ id: 2, senseId: "cpu-usage", payload: "reload-test", ts: Date.now() });
const startThreadCall = mockChildren
.flatMap((c) => (c.send as ReturnType<typeof vi.fn>).mock.calls as [unknown][])
.find(
([msg]) =>
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "start-thread" &&
(msg as Record<string, unknown>).workflow === "new-workflow",
);
expect(startThreadCall).toBeDefined();
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
it("old workflow reflexes are removed after reloadConfig", async () => {
const logStore = makeLogStore();
const initialConfig = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [{ kind: "workflow", workflow: "old-workflow", on: ["cpu-usage"] }],
workflows: { "old-workflow": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(initialConfig, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
// Reload with the workflow reflex removed
const newConfig: NerveConfig = {
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [],
workflows: null,
};
kernel.reloadConfig(newConfig);
// Clear send history
for (const c of mockChildren) {
(c.send as ReturnType<typeof vi.fn>).mockClear();
}
// Emit a signal — old-workflow should NOT be triggered
kernel.bus.emit({ id: 3, senseId: "cpu-usage", payload: "after-reload", ts: Date.now() });
const startThreadCall = mockChildren
.flatMap((c) => (c.send as ReturnType<typeof vi.fn>).mock.calls as [unknown][])
.find(
([msg]) =>
msg !== null &&
typeof msg === "object" &&
(msg as Record<string, unknown>).type === "start-thread",
);
expect(startThreadCall).toBeUndefined();
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
describe("graceful shutdown stops workflow workers", () => {
it("stop() resolves after workflow workers exit", async () => {
const logStore = makeLogStore();
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [{ kind: "workflow", workflow: "shutdown-test", on: ["cpu-usage"] }],
workflows: { "shutdown-test": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
// Trigger a workflow so a worker is spawned
kernel.bus.emit({ id: 1, senseId: "cpu-usage", payload: null, ts: Date.now() });
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await expect(stopPromise).resolves.toBeUndefined();
});
it("workflowManager is exposed on kernel", () => {
const logStore = makeLogStore();
const config = makeConfig({
workflows: { "my-wf": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
expect(kernel.workflowManager).toBeDefined();
expect(typeof kernel.workflowManager.startWorkflow).toBe("function");
expect(typeof kernel.workflowManager.activeCount).toBe("function");
expect(typeof kernel.workflowManager.stop).toBe("function");
kernel.stop().catch(() => {});
});
it("getHealth includes activeWorkflows count", async () => {
const logStore = makeLogStore();
const config = makeConfig({
senses: {
"cpu-usage": { group: "system", throttle: null, timeout: null, gracePeriod: null },
},
reflexes: [{ kind: "workflow", workflow: "health-wf", on: ["cpu-usage"] }],
workflows: { "health-wf": { concurrency: 2, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
workerScript: "fake-worker.js",
logStore,
});
const health = kernel.getHealth();
expect(health).toHaveProperty("activeWorkflows");
expect(typeof health.activeWorkflows).toBe("number");
const stopPromise = kernel.stop();
await vi.runAllTimersAsync();
await stopPromise;
});
});
});
@@ -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({
@@ -0,0 +1,40 @@
import { describe, expect, it } from "vitest";
import {
assertValidUtcDay,
compareIsoDays,
lastArchivableUtcDay,
nextUtcDay,
prevUtcDay,
utcDateStringFromMs,
utcDayEndExclusiveMs,
utcDayStartMs,
} from "../log-archive.js";
describe("log-archive UTC helpers", () => {
it("lastArchivableUtcDay matches RFC-style boundary (exclusive end of day ≤ boundary)", () => {
const boundary = Date.UTC(2026, 1, 2, 12, 0, 0); // 2026-02-02 12:00 UTC
expect(lastArchivableUtcDay(boundary)).toBe("2026-02-01");
});
it("round-trips UTC day bounds", () => {
expect(utcDayStartMs("2026-02-01")).toBe(Date.UTC(2026, 1, 1));
expect(utcDayEndExclusiveMs("2026-02-01")).toBe(Date.UTC(2026, 1, 2));
expect(utcDateStringFromMs(Date.UTC(2026, 1, 1, 23, 59))).toBe("2026-02-01");
});
it("nextUtcDay / prevUtcDay", () => {
expect(nextUtcDay("2026-02-01")).toBe("2026-02-02");
expect(prevUtcDay("2026-02-01")).toBe("2026-01-31");
});
it("compareIsoDays sorts lexicographically for YYYY-MM-DD", () => {
expect(compareIsoDays("2026-01-01", "2026-02-01")).toBeLessThan(0);
expect(compareIsoDays("2026-02-01", "2026-02-01")).toBe(0);
});
it("assertValidUtcDay rejects invalid calendars", () => {
expect(() => assertValidUtcDay("2026-02-31")).toThrow();
expect(() => assertValidUtcDay("bad")).toThrow();
});
});
@@ -0,0 +1,139 @@
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { LOG_ARCHIVE_META_KEY, createLogStore } from "../log-store.js";
import type { LogStore } from "../log-store.js";
const DAY_MS = 86_400_000;
/** `now` such that 2026-02-01 is the last archivable UTC day under a 30-day window. */
function nowForLastArchivableFeb1(): number {
const boundary = Date.UTC(2026, 1, 2, 12, 0, 0);
return boundary + 30 * DAY_MS;
}
describe("LogStore — cold archive (RFC-001 §5.4)", () => {
let tmpDir: string;
let store: LogStore;
beforeEach(() => {
tmpDir = mkdtempSync(join(tmpdir(), "nerve-archive-"));
store = createLogStore(join(tmpDir, "data", "logs.db"));
});
afterEach(() => {
store.close();
rmSync(tmpDir, { recursive: true, force: true });
});
it("exports one UTC day to JSONL, deletes rows, advances archived_up_to", () => {
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "system", type: "x", refId: null, payload: '{"a":1}', ts });
store.append({ source: "reflex", type: "y", refId: "z", payload: null, ts: ts + 1 });
const now = nowForLastArchivableFeb1();
const result = store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
expect(result.days).toHaveLength(1);
expect(result.days[0].day).toBe("2026-02-01");
expect(result.days[0].rowCount).toBe(2);
const jsonlPath = join(tmpDir, "data", "archive", "logs", "2026-02-01.jsonl");
expect(result.days[0].filePath).toBe(jsonlPath);
const lines = readFileSync(jsonlPath, "utf8").trim().split("\n");
expect(lines).toHaveLength(2);
const o = JSON.parse(lines[0] ?? "{}") as Record<string, unknown>;
expect(o.source).toBe("system");
expect(o.refId).toBeNull();
expect(store.query()).toHaveLength(0);
expect(store.getMeta(LOG_ARCHIVE_META_KEY)).toBe("2026-02-01");
});
it("returns nothing for an empty logs table", () => {
const r = store.archiveLogs({ now: nowForLastArchivableFeb1(), retentionMs: 30 * DAY_MS });
expect(r.days).toHaveLength(0);
expect(store.getMeta(LOG_ARCHIVE_META_KEY)).toBeNull();
});
it("does nothing when all logs are inside the hot window", () => {
const now = Date.UTC(2026, 3, 23, 12, 0, 0);
const ts = now - 5 * DAY_MS;
store.append({ source: "system", type: "warm", refId: null, payload: null, ts });
const r = store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
expect(r.days).toHaveLength(0);
expect(store.query()).toHaveLength(1);
});
it("second archive with same clock is a no-op (watermark already caught up)", () => {
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "system", type: "x", refId: null, payload: null, ts });
const now = nowForLastArchivableFeb1();
store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
const path = join(tmpDir, "data", "archive", "logs", "2026-02-01.jsonl");
const first = readFileSync(path, "utf8");
const second = store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
expect(second.days).toHaveLength(0);
expect(readFileSync(path, "utf8")).toBe(first);
});
it("overwrites JSONL when the same UTC day is archived again after watermark rewind", () => {
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "a", type: "1", refId: null, payload: null, ts });
const now = nowForLastArchivableFeb1();
store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
store.setMeta(LOG_ARCHIVE_META_KEY, "2026-01-31");
store.append({ source: "b", type: "2", refId: null, payload: null, ts: ts + 100 });
store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
const path = join(tmpDir, "data", "archive", "logs", "2026-02-01.jsonl");
const lines = readFileSync(path, "utf8").trim().split("\n");
expect(lines).toHaveLength(1);
expect(JSON.parse(lines[0] ?? "{}").source).toBe("b");
});
it("respects maxDays across invocations", () => {
const t1 = Date.UTC(2026, 1, 1, 10, 0, 0);
const t2 = Date.UTC(2026, 1, 2, 10, 0, 0);
store.append({ source: "system", type: "a", refId: null, payload: null, ts: t1 });
store.append({ source: "system", type: "b", refId: null, payload: null, ts: t2 });
const now = Date.UTC(2027, 0, 1, 12, 0, 0);
const r1 = store.archiveLogs({ now, retentionMs: 30 * DAY_MS, maxDays: 1 });
expect(r1.days).toHaveLength(1);
expect(r1.days[0].day).toBe("2026-02-01");
const r2 = store.archiveLogs({ now, retentionMs: 30 * DAY_MS, maxDays: 1 });
expect(r2.days).toHaveLength(1);
expect(r2.days[0].day).toBe("2026-02-02");
expect(store.getMeta(LOG_ARCHIVE_META_KEY)).toBe("2026-02-02");
expect(store.query()).toHaveLength(0);
});
it("starts from earliest log day when it is before watermark+1", () => {
store.setMeta(LOG_ARCHIVE_META_KEY, "2026-01-10");
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "x", type: "p", refId: null, payload: null, ts });
const result = store.archiveLogs({ now: nowForLastArchivableFeb1(), retentionMs: 30 * DAY_MS });
expect(result.days.map((d) => d.day)).toContain("2026-02-01");
});
it("throws on invalid archived_up_to watermark", () => {
store.setMeta(LOG_ARCHIVE_META_KEY, "not-a-date");
expect(() => store.archiveLogs({ now: Date.now() })).toThrow();
});
it("runs VACUUM when vacuum: true", () => {
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "system", type: "x", refId: null, payload: null, ts });
const r = store.archiveLogs({
now: nowForLastArchivableFeb1(),
retentionMs: 30 * DAY_MS,
vacuum: true,
});
expect(r.vacuumed).toBe(true);
});
});
@@ -0,0 +1,198 @@
/**
* Phase 3 — LogStore crash recovery helpers tests.
*
* Tests for getThreadEvents() and getTriggerPayload().
*/
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { createLogStore } from "../log-store.js";
import type { LogStore } from "../log-store.js";
describe("LogStore — crash recovery helpers (Phase 3)", () => {
let tmpDir: string;
let store: LogStore;
beforeEach(() => {
tmpDir = mkdtempSync(join(tmpdir(), "nerve-cr-log-test-"));
store = createLogStore(join(tmpDir, "data", "logs.db"));
});
afterEach(() => {
store.close();
rmSync(tmpDir, { recursive: true, force: true });
});
describe("getTriggerPayload", () => {
it("returns null for an unknown runId", () => {
expect(store.getTriggerPayload("no-such-run")).toBeNull();
});
it("returns the triggerPayload stored in the 'started' log entry", () => {
const payload = { task: "build", repo: "myrepo" };
store.upsertWorkflowRun(
{
source: "workflow",
type: "started",
refId: "run-1",
payload: JSON.stringify({ triggerPayload: payload }),
ts: 1000,
},
{ runId: "run-1", workflow: "my-wf", status: "started", ts: 1000 },
);
const result = store.getTriggerPayload("run-1");
expect(result).toMatchObject(payload);
});
it("returns null when started log entry has no payload", () => {
store.upsertWorkflowRun(
{
source: "workflow",
type: "started",
refId: "run-2",
payload: null,
ts: 1000,
},
{ runId: "run-2", workflow: "my-wf", status: "started", ts: 1000 },
);
expect(store.getTriggerPayload("run-2")).toBeNull();
});
it("returns the payload from the first 'started' entry (earliest)", () => {
const payloadA = { trigger: "first" };
const payloadB = { trigger: "second" };
// Insert two started entries for the same run
store.append({
source: "workflow",
type: "started",
refId: "run-3",
payload: JSON.stringify({ triggerPayload: payloadA }),
ts: 100,
});
store.append({
source: "workflow",
type: "started",
refId: "run-3",
payload: JSON.stringify({ triggerPayload: payloadB }),
ts: 200,
});
const result = store.getTriggerPayload("run-3");
// Should return the first (earliest) started entry
expect(result).toMatchObject(payloadA);
});
});
describe("getThreadEvents", () => {
it("returns empty array for an unknown runId", () => {
expect(store.getThreadEvents("no-such-run")).toHaveLength(0);
});
it("returns CommandEvents in insertion order", () => {
const events = [
{ type: "thread_start", triggerPayload: {} },
{ type: "scan_complete", items: ["a", "b"] },
{ type: "process_done", count: 2 },
];
for (const event of events) {
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-4",
payload: JSON.stringify(event),
ts: Date.now(),
});
}
const result = store.getThreadEvents("run-4");
expect(result).toHaveLength(3);
expect(result[0].type).toBe("thread_start");
expect(result[1].type).toBe("scan_complete");
expect(result[2].type).toBe("process_done");
});
it("skips entries with null payload", () => {
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-5",
payload: null,
ts: 1000,
});
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-5",
payload: JSON.stringify({ type: "valid_event" }),
ts: 1001,
});
const result = store.getThreadEvents("run-5");
expect(result).toHaveLength(1);
expect(result[0].type).toBe("valid_event");
});
it("only returns thread_command_event entries (not other workflow log types)", () => {
// Insert a mix of workflow log types
store.upsertWorkflowRun(
{
source: "workflow",
type: "started",
refId: "run-6",
payload: JSON.stringify({ triggerPayload: {} }),
ts: 1000,
},
{ runId: "run-6", workflow: "my-wf", status: "started", ts: 1000 },
);
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-6",
payload: JSON.stringify({ type: "step_one" }),
ts: 1001,
});
store.append({
source: "workflow",
type: "step_complete",
refId: "run-6",
payload: JSON.stringify({ message: "done step" }),
ts: 1002,
});
const result = store.getThreadEvents("run-6");
expect(result).toHaveLength(1);
expect(result[0].type).toBe("step_one");
});
it("does not return events from a different runId", () => {
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-7",
payload: JSON.stringify({ type: "event_for_7" }),
ts: 1000,
});
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-8",
payload: JSON.stringify({ type: "event_for_8" }),
ts: 1001,
});
const result7 = store.getThreadEvents("run-7");
expect(result7).toHaveLength(1);
expect(result7[0].type).toBe("event_for_7");
const result8 = store.getThreadEvents("run-8");
expect(result8).toHaveLength(1);
expect(result8[0].type).toBe("event_for_8");
});
});
});
@@ -7,6 +7,7 @@ import { drizzle } from "drizzle-orm/better-sqlite3";
import { integer, real, sqliteTable } from "drizzle-orm/sqlite-core";
import { describe, expect, it } from "vitest";
import { createBlobStore } from "../blob-store.js";
import { parseParentMessage } from "../ipc.js";
import { executeCompute, openPeerDb, openSenseDb, runMigrations } from "../sense-runtime.js";
import type { DrizzleDB, PeerMap, SenseRuntime } from "../sense-runtime.js";
@@ -340,6 +341,20 @@ describe("executeCompute", () => {
expect(capturedSignal).toBeInstanceOf(AbortSignal);
sqlite.close();
});
it("passes BlobStore as options.blobs when blobStore argument is provided", async () => {
const blobsRoot = mkdtempSync(join(tmpdir(), "nerve-blobs-"));
const blobStore = createBlobStore(blobsRoot);
let seen: ReturnType<typeof createBlobStore> | undefined;
const { runtime, sqlite } = makeRuntime(async (_db, _peers, options) => {
seen = options?.blobs;
return null;
});
await executeCompute(runtime, emptyPeers, undefined, blobStore);
expect(seen).toBe(blobStore);
sqlite.close();
});
});
// ---------------------------------------------------------------------------
@@ -72,6 +72,9 @@ function makeLogStore() {
appendWithWorkflowUpdate: vi.fn(),
getWorkflowRun: vi.fn(() => null),
getActiveWorkflowRuns: vi.fn(() => []),
getTriggerPayload: vi.fn(() => null),
getThreadEvents: vi.fn(() => []),
archiveLogs: vi.fn(() => ({ days: [], vacuumed: false })),
close: vi.fn(),
};
}
+106
View File
@@ -0,0 +1,106 @@
/**
* CAS blob store — sha256 content-addressable files under `data/blobs/`.
*
* Layout: `<root>/<2-hex-shard>/<62-hex-rest>` (RFC-001 §8).
*/
import { createHash, randomBytes } from "node:crypto";
import {
existsSync,
mkdirSync,
readFileSync,
renameSync,
unlinkSync,
writeFileSync,
} from "node:fs";
import { dirname, join } from "node:path";
const SHA256_HEX_LEN = 64;
const HEX_RE = /^[0-9a-f]+$/;
export type BlobStore = {
/** Persist UTF-8 or raw bytes; returns lowercase hex sha256. Idempotent for identical content. */
write: (content: string | Uint8Array | Buffer) => string;
/** Returns bytes or null if the hash is invalid or no blob exists. Verifies digest matches path. */
read: (hash: string) => Buffer | null;
/** True when hash is well-formed and the blob file is present. */
exists: (hash: string) => boolean;
};
function toBuffer(content: string | Uint8Array | Buffer): Buffer {
if (typeof content === "string") return Buffer.from(content, "utf8");
if (Buffer.isBuffer(content)) return content;
return Buffer.from(content);
}
function digestHex(buf: Buffer): string {
return createHash("sha256").update(buf).digest("hex");
}
/** @returns normalized lowercase hex or null if not a valid sha256 hex string */
export function normalizeBlobHash(hash: string): string | null {
const h = hash.trim().toLowerCase();
if (h.length !== SHA256_HEX_LEN) return null;
if (!HEX_RE.test(h)) return null;
return h;
}
function pathForHash(blobsRoot: string, hashLower: string): string {
return join(blobsRoot, hashLower.slice(0, 2), hashLower.slice(2));
}
function verifyPathMatchesContent(filePath: string, expectedHash: string): Buffer {
const data = readFileSync(filePath);
const actual = digestHex(data);
if (actual !== expectedHash) {
throw new Error(
`Blob CAS mismatch at "${filePath}": file digests to ${actual}, path expects ${expectedHash}`,
);
}
return data;
}
export function createBlobStore(blobsRoot: string): BlobStore {
function write(content: string | Uint8Array | Buffer): string {
const buf = toBuffer(content);
const hash = digestHex(buf);
const filePath = pathForHash(blobsRoot, hash);
if (existsSync(filePath)) {
verifyPathMatchesContent(filePath, hash);
return hash;
}
mkdirSync(dirname(filePath), { recursive: true });
const tmp = join(dirname(filePath), `.tmp.${randomBytes(16).toString("hex")}`);
try {
writeFileSync(tmp, buf);
renameSync(tmp, filePath);
} catch (e) {
try {
unlinkSync(tmp);
} catch {
// ignore cleanup errors
}
throw e;
}
return hash;
}
function read(hash: string): Buffer | null {
const h = normalizeBlobHash(hash);
if (h === null) return null;
const filePath = pathForHash(blobsRoot, h);
if (!existsSync(filePath)) return null;
return verifyPathMatchesContent(filePath, h);
}
function exists(hash: string): boolean {
const h = normalizeBlobHash(hash);
if (h === null) return false;
return existsSync(pathForHash(blobsRoot, h));
}
return { write, read, exists };
}
+162
View File
@@ -0,0 +1,162 @@
/**
* 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)
*/
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";
workflow: string;
payload: unknown;
};
/** 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 = JSON.parse(line) as unknown;
if (obj === null || typeof obj !== "object") return null;
const req = obj as Record<string, unknown>;
if (req.type === "trigger-workflow") {
if (typeof req.workflow !== "string" || req.workflow.length === 0) return null;
return { type: "trigger-workflow", workflow: req.workflow, payload: req.payload ?? {} };
}
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;
/** Called when a list-senses request arrives. Returns sense info for all registered senses. */
listSenses: () => SenseInfo[];
};
export function createDaemonIpcServer(
socketPath: string,
workflowManager: WorkflowManager,
opts: DaemonIpcServerOptions,
): DaemonIpcServer {
// Remove stale socket file if it exists
try {
rmSync(socketPath);
} catch {
// file did not exist — that is fine
}
function handleLine(socket: Socket, line: string): void {
const trimmed = line.trim();
if (trimmed.length === 0) return;
const req = parseRequest(trimmed);
if (req === null) {
const resp: DaemonResponse = { ok: false, error: "Invalid request" };
socket.write(`${JSON.stringify(resp)}\n`);
return;
}
try {
if (req.type === "trigger-workflow") {
workflowManager.startWorkflow(req.workflow, req.payload);
const resp: DaemonResponse = { ok: true };
socket.write(`${JSON.stringify(resp)}\n`);
} else if (req.type === "trigger-sense") {
opts.triggerSense(req.sense);
const resp: DaemonResponse = { ok: true };
socket.write(`${JSON.stringify(resp)}\n`);
} else if (req.type === "list-senses") {
const senses = opts.listSenses();
const resp: DaemonResponse = { ok: true, senses };
socket.write(`${JSON.stringify(resp)}\n`);
}
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
const resp: DaemonResponse = { ok: false, error: msg };
socket.write(`${JSON.stringify(resp)}\n`);
}
}
const server: Server = createServer((socket) => {
let buf = "";
socket.on("data", (chunk: Buffer) => {
buf += chunk.toString("utf8");
const lines = buf.split("\n");
buf = lines.pop() ?? "";
for (const line of lines) {
handleLine(socket, line);
}
});
socket.on("error", () => {
// client disconnected mid-message — ignore
});
});
server.listen(socketPath, () => {
process.stderr.write(`[daemon-ipc] listening on ${socketPath}\n`);
});
server.on("error", (err) => {
process.stderr.write(`[daemon-ipc] server error: ${err.message}\n`);
});
async function close(): Promise<void> {
await new Promise<void>((resolve) => {
server.close(() => resolve());
});
try {
rmSync(socketPath);
} catch {
// already removed
}
}
return { close };
}
+31 -14
View File
@@ -31,7 +31,13 @@ export type ConfigFileChange = {
filePath: string;
};
export type FileChange = SenseFileChange | ConfigFileChange;
export type WorkflowFileChange = {
kind: "workflow";
workflowName: string;
filePath: string;
};
export type FileChange = SenseFileChange | ConfigFileChange | WorkflowFileChange;
export type FileChangeHandler = (change: FileChange) => void;
@@ -61,6 +67,28 @@ export function createFileWatcher(
);
}
function handleSenseChange(normalized: string, filename: string): void {
if (!(normalized.startsWith("senses/") && normalized.endsWith(".ts"))) return;
const rel = relative("senses", normalized);
const senseName = rel.split("/")[0];
if (senseName) {
debounced(`sense:${senseName}`, () => {
handler({ kind: "sense", senseName, filePath: join(nerveRoot, filename) });
});
}
}
function handleWorkflowChange(normalized: string, filename: string): void {
if (!(normalized.startsWith("workflows/") && normalized.endsWith(".ts"))) return;
const rel = relative("workflows", normalized);
const workflowName = rel.split("/")[0];
if (workflowName) {
debounced(`workflow:${workflowName}`, () => {
handler({ kind: "workflow", workflowName, filePath: join(nerveRoot, filename) });
});
}
}
function handleFsEvent(_eventType: string, filename: string | null): void {
if (filename === null) return;
@@ -73,19 +101,8 @@ export function createFileWatcher(
return;
}
if (normalized.startsWith("senses/") && normalized.endsWith(".ts")) {
const rel = relative("senses", normalized);
const senseName = rel.split("/")[0];
if (senseName) {
debounced(`sense:${senseName}`, () => {
handler({
kind: "sense",
senseName,
filePath: join(nerveRoot, filename),
});
});
}
}
handleSenseChange(normalized, filename);
handleWorkflowChange(normalized, filename);
}
try {
+16 -2
View File
@@ -29,11 +29,25 @@ 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";
export { createLogStore } from "./log-store.js";
export type { LogStore, LogEntry, LogQuery, WorkflowRun, WorkflowRunStatus } from "./log-store.js";
export { createBlobStore, normalizeBlobHash } from "./blob-store.js";
export type { BlobStore } from "./blob-store.js";
export { createLogStore, LOG_ARCHIVE_META_KEY } from "./log-store.js";
export type {
LogStore,
LogEntry,
LogQuery,
WorkflowRun,
WorkflowRunStatus,
ArchiveLogsDayResult,
ArchiveLogsOptions,
ArchiveLogsResult,
} from "./log-store.js";
export { createWorkflowManager } from "./workflow-manager.js";
export type { WorkflowManager } from "./workflow-manager.js";
+54 -10
View File
@@ -39,6 +39,10 @@ export type StartThreadMessage = {
export type ResumeThreadMessage = {
type: "resume-thread";
runId: string;
/** Serialised CommandEvent history to rebuild ThreadState. */
events: Array<{ type: string; [key: string]: unknown }>;
/** Serialised trigger payload (the same value as in the original start-thread). */
triggerPayload: unknown;
};
/** Union of all messages the parent sends to a worker */
@@ -99,6 +103,14 @@ export type WorkflowErrorMessage = {
error: string;
};
/** Workflow Worker → Parent: a thread CommandEvent produced by a role (for crash recovery). */
export type ThreadCommandEventMessage = {
type: "thread-command-event";
runId: string;
/** The CommandEvent returned by role.execute() — will be persisted for crash recovery. */
event: { type: string; [key: string]: unknown };
};
/** Union of all messages a worker sends to the parent */
export type WorkerToParentMessage =
| SignalMessage
@@ -106,7 +118,8 @@ export type WorkerToParentMessage =
| ReadyMessage
| HealthResponseMessage
| ThreadEventMessage
| WorkflowErrorMessage;
| WorkflowErrorMessage
| ThreadCommandEventMessage;
const PARENT_MSG_TYPES = new Set([
"compute",
@@ -116,6 +129,20 @@ const PARENT_MSG_TYPES = new Set([
"resume-thread",
]);
function validateStartThreadMsg(obj: Record<string, unknown>): string | null {
if (typeof obj.runId !== "string") return "'start-thread' message missing string 'runId'";
if (typeof obj.workflow !== "string") return "'start-thread' message missing string 'workflow'";
if (!("triggerPayload" in obj)) return "'start-thread' message missing 'triggerPayload'";
return null;
}
function validateResumeThreadMsg(obj: Record<string, unknown>): string | null {
if (typeof obj.runId !== "string") return "'resume-thread' message missing string 'runId'";
if (!Array.isArray(obj.events)) return "'resume-thread' message missing 'events' array";
if (!("triggerPayload" in obj)) return "'resume-thread' message missing 'triggerPayload'";
return null;
}
/** Validate and parse an unknown IPC message received from the parent process. */
export function parseParentMessage(raw: unknown): Result<ParentToWorkerMessage> {
if (raw === null || typeof raw !== "object") {
@@ -129,16 +156,12 @@ export function parseParentMessage(raw: unknown): Result<ParentToWorkerMessage>
return err(new Error(`Unknown IPC message type: "${obj.type}"`));
}
if (obj.type === "start-thread") {
if (typeof obj.runId !== "string")
return err(new Error("'start-thread' message missing string 'runId'"));
if (typeof obj.workflow !== "string")
return err(new Error("'start-thread' message missing string 'workflow'"));
if (!("triggerPayload" in obj))
return err(new Error("'start-thread' message missing 'triggerPayload'"));
const errMsg = validateStartThreadMsg(obj);
if (errMsg !== null) return err(new Error(errMsg));
}
if (obj.type === "resume-thread") {
if (typeof obj.runId !== "string")
return err(new Error("'resume-thread' message missing string 'runId'"));
const errMsg = validateResumeThreadMsg(obj);
if (errMsg !== null) return err(new Error(errMsg));
}
return ok(raw as ParentToWorkerMessage);
}
@@ -192,7 +215,9 @@ function parseThreadEventMsg(
return err(new Error("Worker 'thread-event' message missing string 'runId' field"));
}
if (typeof obj.eventType !== "string" || !THREAD_EVENT_TYPES.has(obj.eventType)) {
return err(new Error(`Worker 'thread-event' message has invalid 'eventType': "${obj.eventType}"`));
return err(
new Error(`Worker 'thread-event' message has invalid 'eventType': "${obj.eventType}"`),
);
}
if (!("payload" in obj)) {
return err(new Error("Worker 'thread-event' message missing 'payload' field"));
@@ -220,8 +245,26 @@ const WORKER_MSG_TYPES = new Set([
"health-response",
"thread-event",
"workflow-error",
"thread-command-event",
]);
function parseThreadCommandEventMsg(
obj: Record<string, unknown>,
raw: unknown,
): Result<WorkerToParentMessage> {
if (typeof obj.runId !== "string") {
return err(new Error("Worker 'thread-command-event' message missing string 'runId' field"));
}
if (obj.event === null || typeof obj.event !== "object") {
return err(new Error("Worker 'thread-command-event' message missing object 'event' field"));
}
const event = obj.event as Record<string, unknown>;
if (typeof event.type !== "string") {
return err(new Error("Worker 'thread-command-event' event missing string 'type' field"));
}
return ok(raw as ThreadCommandEventMessage);
}
/** Validate and parse an unknown IPC message received from a worker process. */
export function parseWorkerMessage(raw: unknown): Result<WorkerToParentMessage> {
if (raw === null || typeof raw !== "object") {
@@ -239,5 +282,6 @@ export function parseWorkerMessage(raw: unknown): Result<WorkerToParentMessage>
if (obj.type === "health-response") return parseHealthResponseMsg(obj, raw);
if (obj.type === "thread-event") return parseThreadEventMsg(obj, raw);
if (obj.type === "workflow-error") return parseWorkflowErrorMsg(obj, raw);
if (obj.type === "thread-command-event") return parseThreadCommandEventMsg(obj, raw);
return ok({ type: "ready" });
}
+126 -8
View File
@@ -18,9 +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 } from "./daemon-ipc.js";
import { createFileWatcher } from "./file-watcher.js";
import type { FileWatcher } from "./file-watcher.js";
import type { ComputeMessage, ShutdownMessage } from "./ipc.js";
@@ -31,12 +33,15 @@ import { createReflexScheduler } from "./reflex-scheduler.js";
import type { ReflexScheduler } from "./reflex-scheduler.js";
import { createSignalBus } from "./signal-bus.js";
import type { SignalBus } from "./signal-bus.js";
import { createWorkflowManager } from "./workflow-manager.js";
import type { WorkflowManager } from "./workflow-manager.js";
export type KernelHealth = {
uptime: number;
activeSenses: number;
activeGroups: number;
pendingComputes: number;
activeWorkflows: number;
memoryUsage: NodeJS.MemoryUsage;
};
@@ -46,12 +51,18 @@ export type Kernel = {
senseCount: number;
bus: SignalBus;
logStore: LogStore;
workflowManager: WorkflowManager;
/** Resolves when all workers have sent their initial "ready" message. */
ready: Promise<void>;
/** Returns the PID of the worker process for a given group, or null if not found. */
getWorkerPid: (group: string) => number | null;
/** Sends a compute message to the worker responsible for the given sense. */
triggerCompute: (senseName: string) => void;
/**
* On-demand sense trigger — looks up the group for `senseName`, finds its worker,
* and sends a compute message. Throws if the sense is unknown.
*/
triggerSense: (senseName: string) => void;
/** Gracefully restart a group worker (wait for exit, then respawn). */
restartGroup: (group: string) => Promise<void>;
/** Reload config from a new NerveConfig, incrementally updating scheduler and workers.
@@ -74,9 +85,16 @@ function resolveWorkerScript(): string {
}
function spawnWorker(nerveRoot: string, group: string, workerScript: string): ChildProcess {
return fork(workerScript, ["--group", group, "--root", nerveRoot], {
const child = fork(workerScript, ["--group", group, "--root", nerveRoot], {
stdio: ["ignore", "inherit", "inherit", "ipc"],
});
// Prevent unhandled EPIPE when writing to a child whose IPC channel closed
child.on("error", (err) => {
if ((err as NodeJS.ErrnoException).code !== "EPIPE") {
console.error("[worker] error:", err.message);
}
});
return child;
}
function sendCompute(worker: ChildProcess, senseName: string): void {
@@ -107,10 +125,15 @@ function groupForSense(config: NerveConfig, senseName: string): string | null {
}
export type KernelOptions = {
workerScript: string;
workerScript?: string | null;
enableFileWatcher?: boolean;
/** Override the LogStore instance (useful for testing). */
logStore?: LogStore;
/**
* Unix socket path for the daemon IPC server (used by CLI to send trigger-workflow).
* When null, the IPC server is not started (e.g. during tests).
*/
ipcSocketPath?: string | null;
};
function defaultKernelOptions(): KernelOptions {
@@ -123,7 +146,7 @@ export function createKernel(
options: KernelOptions = defaultKernelOptions(),
): Kernel {
const bus: SignalBus = createSignalBus();
const workerScript = options.workerScript;
const workerScript = options.workerScript ?? resolveWorkerScript();
const startTime = Date.now();
const logStore: LogStore = options.logStore ?? createLogStore(join(nerveRoot, "data", "logs.db"));
@@ -143,6 +166,8 @@ export function createKernel(
return _signalIdCounter;
}
const workflowManager = createWorkflowManager(nerveRoot, config, logStore);
const groups = new Set<string>();
for (const senseConfig of Object.values(config.senses)) {
groups.add(senseConfig.group);
@@ -201,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,
@@ -267,7 +292,24 @@ export function createKernel(
sendCompute(entry.process, senseName);
}
scheduler = createReflexScheduler(config, bus, triggerFn, { logStore });
function triggerSense(senseName: string): void {
const group = groupForSense(config, senseName);
if (group === null) {
throw new Error(`Unknown sense: "${senseName}"`);
}
const entry = workers.get(group);
if (entry === undefined) {
throw new Error(`No worker running for group "${group}" (sense: "${senseName}")`);
}
sendCompute(entry.process, senseName);
}
scheduler = createReflexScheduler(config, bus, triggerFn, {
logStore,
workflowTriggerFn: (workflowName, payload) => {
workflowManager.startWorkflow(workflowName, payload);
},
});
if (groups.size === 0) {
readyResolve?.();
@@ -346,11 +388,37 @@ export function createKernel(
function reloadConfig(newConfig: NerveConfig): void {
const oldGroups = collectGroups(config);
const oldConfig = config;
const oldWorkflows = config.workflows ?? {};
config = newConfig;
// Note: pending/throttled computes in the old scheduler are silently dropped here.
// In-flight state is not preserved across reloadConfig.
scheduler.stop();
scheduler = createReflexScheduler(config, bus, triggerFn, { logStore });
scheduler = createReflexScheduler(config, bus, triggerFn, {
logStore,
workflowTriggerFn: (workflowName, payload) => {
workflowManager.startWorkflow(workflowName, payload);
},
});
// Update workflow concurrency/overflow config incrementally — no restart needed
workflowManager.updateConfig(newConfig);
const newWorkflows = newConfig.workflows ?? {};
// Drain + remove workers for deleted workflows
for (const workflowName of Object.keys(oldWorkflows)) {
if (!(workflowName in newWorkflows)) {
process.stderr.write(
`[kernel] workflow "${workflowName}" removed from config, draining worker\n`,
);
workflowManager.drainAndRespawn(workflowName).catch((e) => {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(
`[kernel] drainAndRespawn error for removed workflow "${workflowName}": ${msg}\n`,
);
});
}
}
const newGroups = collectGroups(newConfig);
removeStaleGroups(oldGroups, newGroups);
addNewGroups(oldGroups, newGroups);
@@ -378,6 +446,7 @@ export function createKernel(
activeSenses: Object.keys(config.senses).length,
activeGroups: workers.size,
pendingComputes: 0,
activeWorkflows: workflowManager.totalActiveCount(),
memoryUsage: process.memoryUsage(),
};
}
@@ -401,6 +470,23 @@ export function createKernel(
});
}
function handleWorkflowFileChange(workflowName: string): void {
process.stderr.write(
`[kernel] workflow file changed: "${workflowName}", draining and respawning worker\n`,
);
logStore.append({
source: "system",
type: "workflow_reload",
refId: workflowName,
payload: null,
ts: Date.now(),
});
workflowManager.drainAndRespawn(workflowName).catch((e) => {
const msg = e instanceof Error ? e.message : String(e);
process.stderr.write(`[kernel] drainAndRespawn error for "${workflowName}": ${msg}\n`);
});
}
function handleConfigFileChange(): void {
process.stderr.write("[kernel] nerve.yaml changed, reloading config\n");
logStore.append({
@@ -431,6 +517,31 @@ export function createKernel(
fileWatcher = createFileWatcher(nerveRoot, (change) => {
if (change.kind === "sense") handleSenseFileChange(change.senseName);
if (change.kind === "config") handleConfigFileChange();
if (change.kind === "workflow") handleWorkflowFileChange(change.workflowName);
});
}
let ipcServer: DaemonIpcServer | null = null;
if (options.ipcSocketPath != null) {
ipcServer = createDaemonIpcServer(options.ipcSocketPath, workflowManager, {
triggerSense,
listSenses(): SenseInfo[] {
return Object.entries(config.senses).map(([name, senseConfig]) => {
const entries = logStore.query({
source: "sense",
type: "signal",
refId: name,
});
const lastEntry = entries.length > 0 ? entries[entries.length - 1] : null;
return {
name,
group: senseConfig.group,
throttle: senseConfig.throttle,
timeout: senseConfig.timeout,
lastSignalTs: lastEntry !== null ? lastEntry.ts : null,
};
});
},
});
}
@@ -440,7 +551,12 @@ export function createKernel(
fileWatcher.close();
fileWatcher = null;
}
if (ipcServer !== null) {
await ipcServer.close();
ipcServer = null;
}
scheduler.stop();
await workflowManager.stop();
const exitPromises: Promise<void>[] = [];
for (const entry of workers.values()) {
sendShutdown(entry.process);
@@ -469,9 +585,11 @@ export function createKernel(
senseCount,
bus,
logStore,
workflowManager,
ready,
getWorkerPid,
triggerCompute: triggerFn,
triggerSense,
restartGroup,
reloadConfig,
getHealth,
+78
View File
@@ -0,0 +1,78 @@
/** Log cold-archive helpers (RFC-001 §5.4) — UTC calendar days, JSONL export. */
export const LOG_ARCHIVE_META_KEY = "archived_up_to";
export const DEFAULT_LOG_RETENTION_MS = 30 * 86_400_000;
export type ArchiveLogsOptions = {
/** Wall clock for retention boundary (default: `Date.now()`). */
now?: number;
/** Run `VACUUM` after archiving (outside the per-day transaction). */
vacuum?: boolean;
/** Max UTC days to process in one call (default: unlimited). */
maxDays?: number;
/** Override default 30-day retention (tests). */
retentionMs?: number;
};
export type ArchiveLogsDayResult = {
day: string;
rowCount: number;
filePath: string;
};
export type ArchiveLogsResult = {
days: ArchiveLogsDayResult[];
vacuumed: boolean;
};
export function utcDateStringFromMs(ms: number): string {
return new Date(ms).toISOString().slice(0, 10);
}
function parseUtcDayParts(day: string): [number, number, number] {
const m = /^(\d{4})-(\d{2})-(\d{2})$/.exec(day);
if (m === null) {
throw new Error(`Invalid UTC day (expected YYYY-MM-DD): ${day}`);
}
const y = Number(m[1]);
const mo = Number(m[2]);
const d = Number(m[3]);
const t = Date.UTC(y, mo - 1, d);
if (utcDateStringFromMs(t) !== day) {
throw new Error(`Invalid UTC calendar day: ${day}`);
}
return [y, mo, d];
}
export function assertValidUtcDay(day: string): void {
parseUtcDayParts(day);
}
export function utcDayStartMs(day: string): number {
const [y, mo, d] = parseUtcDayParts(day);
return Date.UTC(y, mo - 1, d);
}
export function utcDayEndExclusiveMs(day: string): number {
return utcDayStartMs(day) + 86_400_000;
}
export function prevUtcDay(day: string): string {
return utcDateStringFromMs(utcDayStartMs(day) - 86_400_000);
}
export function nextUtcDay(day: string): string {
return utcDateStringFromMs(utcDayEndExclusiveMs(day));
}
/** Last UTC calendar day D such that the exclusive end of D is ≤ boundaryMs. */
export function lastArchivableUtcDay(boundaryMs: number): string {
return prevUtcDay(utcDateStringFromMs(boundaryMs));
}
export function compareIsoDays(a: string, b: string): number {
if (a < b) return -1;
if (a > b) return 1;
return 0;
}
+251 -12
View File
@@ -7,11 +7,27 @@
* Also provides a `meta` key-value table for bookkeeping (e.g. archive watermarks).
*/
import { mkdirSync } from "node:fs";
import { dirname } from "node:path";
import { mkdirSync, writeFileSync } from "node:fs";
import { dirname, join } from "node:path";
import Database from "better-sqlite3";
import type BetterSqlite3 from "better-sqlite3";
import {
DEFAULT_LOG_RETENTION_MS,
LOG_ARCHIVE_META_KEY,
assertValidUtcDay,
compareIsoDays,
lastArchivableUtcDay,
nextUtcDay,
utcDateStringFromMs,
utcDayEndExclusiveMs,
utcDayStartMs,
} from "./log-archive.js";
import type { ArchiveLogsDayResult, ArchiveLogsOptions, ArchiveLogsResult } from "./log-archive.js";
export { LOG_ARCHIVE_META_KEY } from "./log-archive.js";
export type { ArchiveLogsDayResult, ArchiveLogsOptions, ArchiveLogsResult } from "./log-archive.js";
export type LogEntry = {
id?: number;
source: string;
@@ -40,7 +56,8 @@ export type WorkflowRunStatus =
| "completed"
| "failed"
| "crashed"
| "dropped";
| "dropped"
| "interrupted";
const VALID_WORKFLOW_STATUSES = new Set<string>([
"queued",
@@ -49,6 +66,7 @@ const VALID_WORKFLOW_STATUSES = new Set<string>([
"failed",
"crashed",
"dropped",
"interrupted",
]);
function validateWorkflowRunStatus(status: string): WorkflowRunStatus {
@@ -75,18 +93,12 @@ export type LogStore = {
* Append a workflow log event and atomically upsert the workflow_runs
* materialized table — both in a single SQLite transaction (RFC-002 §6.2).
*/
upsertWorkflowRun: (
entry: Omit<LogEntry, "id">,
run: WorkflowRun,
) => LogEntry;
upsertWorkflowRun: (entry: Omit<LogEntry, "id">, run: WorkflowRun) => LogEntry;
/**
* Alias for upsertWorkflowRun — append a log entry and update workflow_runs
* in one atomic transaction.
*/
appendWithWorkflowUpdate: (
entry: Omit<LogEntry, "id">,
run: WorkflowRun,
) => LogEntry;
appendWithWorkflowUpdate: (entry: Omit<LogEntry, "id">, run: WorkflowRun) => LogEntry;
/** Get the current materialized state of a specific workflow run. */
getWorkflowRun: (runId: string) => WorkflowRun | null;
/**
@@ -94,6 +106,27 @@ export type LogStore = {
* Optionally filter by workflow name.
*/
getActiveWorkflowRuns: (workflowName?: string) => WorkflowRun[];
/**
* Get all workflow runs regardless of status, sorted by ts descending.
* Optionally filter by workflow name.
*/
getAllWorkflowRuns: (workflowName: string | null) => WorkflowRun[];
/**
* Get the trigger payload for a workflow run (stored in the 'started' log entry).
* Returns null if not found.
*/
getTriggerPayload: (runId: string) => unknown;
/**
* Get all workflow CommandEvents for a specific run, ordered by id ASC.
* Used for crash recovery to rebuild ThreadState.
*/
getThreadEvents: (runId: string) => Array<{ type: string; [key: string]: unknown }>;
/**
* Export logs older than the retention window to `data/archive/logs/YYYY-MM-DD.jsonl`,
* then delete those rows and advance `meta.archived_up_to` in one transaction per day
* (RFC-001 §5.4).
*/
archiveLogs: (options?: ArchiveLogsOptions) => ArchiveLogsResult;
close: () => void;
};
@@ -127,6 +160,78 @@ CREATE INDEX IF NOT EXISTS idx_workflow_runs_status ON workflow_runs(status);
CREATE INDEX IF NOT EXISTS idx_workflow_runs_workflow ON workflow_runs(workflow);
`;
type SqlLogRow = {
id: number;
source: string;
type: string;
ref_id: string | null;
payload: string | null;
ts: number;
};
function buildJsonlBody(rows: SqlLogRow[]): string {
if (rows.length === 0) return "";
const lines = rows.map((r) =>
JSON.stringify({
id: r.id,
source: r.source,
type: r.type,
refId: r.ref_id,
payload: r.payload,
ts: r.ts,
}),
);
return `${lines.join("\n")}\n`;
}
function runOptionalVacuum(sqlite: BetterSqlite3.Database, vacuum?: boolean): boolean {
if (vacuum !== true) return false;
sqlite.exec("VACUUM");
return true;
}
function resolveArchiveStartDay(watermark: string | null, minDay: string): string {
if (watermark === null) return minDay;
const afterWatermark = nextUtcDay(watermark);
return compareIsoDays(minDay, afterWatermark) < 0 ? minDay : afterWatermark;
}
function runArchiveDayLoop(
dbPath: string,
options: ArchiveLogsOptions,
selectLogsForDayStmt: BetterSqlite3.Statement,
archiveDayTx: (day: string, start: number, endExclusive: number) => void,
startDay: string,
lastDay: string,
): ArchiveLogsDayResult[] {
const archiveDir = join(dirname(dbPath), "archive", "logs");
mkdirSync(archiveDir, { recursive: true });
const days: ArchiveLogsDayResult[] = [];
let d = startDay;
let processed = 0;
while (compareIsoDays(d, lastDay) <= 0) {
if (options.maxDays !== undefined && processed >= options.maxDays) {
break;
}
const start = utcDayStartMs(d);
const endExclusive = utcDayEndExclusiveMs(d);
const rows = selectLogsForDayStmt.all({ start, endExclusive }) as SqlLogRow[];
const filePath = join(archiveDir, `${d}.jsonl`);
writeFileSync(filePath, buildJsonlBody(rows), "utf8");
archiveDayTx(d, start, endExclusive);
days.push({ day: d, rowCount: rows.length, filePath });
processed += 1;
d = nextUtcDay(d);
}
return days;
}
export function createLogStore(dbPath: string): LogStore {
mkdirSync(dirname(dbPath), { recursive: true });
@@ -151,6 +256,14 @@ export function createLogStore(dbPath: string): LogStore {
"SELECT run_id, workflow, status, ts FROM workflow_runs WHERE run_id = ?",
);
const getTriggerPayloadStmt = sqlite.prepare(
"SELECT payload FROM logs WHERE source = 'workflow' AND type = 'started' AND ref_id = ? ORDER BY id ASC LIMIT 1",
);
const getThreadEventsStmt = sqlite.prepare(
"SELECT payload FROM logs WHERE source = 'workflow' AND type = 'thread_command_event' AND ref_id = ? ORDER BY id ASC",
);
const getActiveWorkflowRunsStmt = sqlite.prepare(
"SELECT run_id, workflow, status, ts FROM workflow_runs WHERE status IN ('queued', 'started') ORDER BY ts ASC",
);
@@ -159,6 +272,22 @@ export function createLogStore(dbPath: string): LogStore {
"SELECT run_id, workflow, status, ts FROM workflow_runs WHERE status IN ('queued', 'started') AND workflow = ? ORDER BY ts ASC",
);
const getAllWorkflowRunsStmt = sqlite.prepare(
"SELECT run_id, workflow, status, ts FROM workflow_runs ORDER BY ts DESC",
);
const getAllWorkflowRunsByNameStmt = sqlite.prepare(
"SELECT run_id, workflow, status, ts FROM workflow_runs WHERE workflow = ? ORDER BY ts DESC",
);
const minLogTsStmt = sqlite.prepare("SELECT MIN(ts) AS m FROM logs");
const selectLogsForDayStmt = sqlite.prepare(
"SELECT id, source, type, ref_id, payload, ts FROM logs WHERE ts >= @start AND ts < @endExclusive ORDER BY id ASC",
);
const deleteLogsForDayStmt = sqlite.prepare(
"DELETE FROM logs WHERE ts >= @start AND ts < @endExclusive",
);
const upsertWorkflowRunTx = sqlite.transaction(
(entry: Omit<LogEntry, "id">, run: WorkflowRun) => {
const info = insertStmt.run({
@@ -281,9 +410,119 @@ export function createLogStore(dbPath: string): LogStore {
}));
}
function getAllWorkflowRuns(workflowName: string | null): WorkflowRun[] {
const rows = (
workflowName !== null
? getAllWorkflowRunsByNameStmt.all(workflowName)
: getAllWorkflowRunsStmt.all()
) as Array<{ run_id: string; workflow: string; status: string; ts: number }>;
return rows.map((r) => ({
runId: r.run_id,
workflow: r.workflow,
status: validateWorkflowRunStatus(r.status),
ts: r.ts,
}));
}
function getTriggerPayload(runId: string): unknown {
const row = getTriggerPayloadStmt.get(runId) as { payload: string | null } | undefined;
if (row === undefined || row.payload === null) return null;
try {
const parsed = JSON.parse(row.payload) as unknown;
if (parsed !== null && typeof parsed === "object") {
const obj = parsed as Record<string, unknown>;
return obj.triggerPayload ?? null;
}
} catch {
// malformed
}
return null;
}
function getThreadEvents(runId: string): Array<{ type: string; [key: string]: unknown }> {
const rows = getThreadEventsStmt.all(runId) as Array<{ payload: string | null }>;
const result: Array<{ type: string; [key: string]: unknown }> = [];
for (const row of rows) {
if (row.payload === null) continue;
try {
const parsed = JSON.parse(row.payload) as unknown;
if (
parsed !== null &&
typeof parsed === "object" &&
typeof (parsed as Record<string, unknown>).type === "string"
) {
result.push(parsed as { type: string; [key: string]: unknown });
}
} catch {
// skip malformed payloads
}
}
return result;
}
const archiveDayTx = sqlite.transaction((day: string, start: number, endExclusive: number) => {
deleteLogsForDayStmt.run({ start, endExclusive });
setMetaStmt.run({ key: LOG_ARCHIVE_META_KEY, value: day });
});
function readWatermark(): string | null {
const raw = getMeta(LOG_ARCHIVE_META_KEY);
if (raw === null) return null;
assertValidUtcDay(raw);
return raw;
}
function firstLogUtcDay(): string | null {
const row = minLogTsStmt.get() as { m: number | null } | undefined;
const m = row?.m;
if (m === null || m === undefined) return null;
return utcDateStringFromMs(m);
}
function archiveLogs(options: ArchiveLogsOptions = {}): ArchiveLogsResult {
const now = options.now ?? Date.now();
const retentionMs = options.retentionMs ?? DEFAULT_LOG_RETENTION_MS;
const lastDay = lastArchivableUtcDay(now - retentionMs);
const watermark = readWatermark();
const minDay = firstLogUtcDay();
if (minDay === null) {
return { days: [], vacuumed: runOptionalVacuum(sqlite, options.vacuum) };
}
const startDay = resolveArchiveStartDay(watermark, minDay);
if (compareIsoDays(startDay, lastDay) > 0) {
return { days: [], vacuumed: runOptionalVacuum(sqlite, options.vacuum) };
}
const days = runArchiveDayLoop(
dbPath,
options,
selectLogsForDayStmt,
archiveDayTx,
startDay,
lastDay,
);
return { days, vacuumed: runOptionalVacuum(sqlite, options.vacuum) };
}
function close(): void {
sqlite.close();
}
return { append, query, getMeta, setMeta, upsertWorkflowRun, appendWithWorkflowUpdate, getWorkflowRun, getActiveWorkflowRuns, close };
return {
append,
query,
getMeta,
setMeta,
upsertWorkflowRun,
appendWithWorkflowUpdate,
getWorkflowRun,
getActiveWorkflowRuns,
getAllWorkflowRuns,
getTriggerPayload,
getThreadEvents,
archiveLogs,
close,
};
}
+19
View File
@@ -16,6 +16,9 @@ import type { SignalBus, Unsubscribe } from "./signal-bus.js";
/** Sends a compute message to the worker responsible for the given sense. */
export type TriggerFn = (senseName: string) => void;
/** Triggers a workflow run in response to a signal. */
export type WorkflowTriggerFn = (workflowName: string, payload: unknown) => void;
/** Per-sense mutable state tracked by the scheduler. */
type SenseState = {
lastComputeAt: number;
@@ -37,6 +40,7 @@ function makeSenseState(): SenseState {
export type ReflexSchedulerOptions = {
logStore?: LogStore;
workflowTriggerFn?: WorkflowTriggerFn;
};
/**
@@ -153,6 +157,21 @@ export function createReflexScheduler(
}
for (const reflex of config.reflexes) {
if (reflex.kind === "workflow") {
if (opts?.workflowTriggerFn !== undefined && reflex.on !== null && reflex.on.length > 0) {
const workflowTriggerFn = opts.workflowTriggerFn;
const workflowName = reflex.workflow;
const watchedSenses = new Set(reflex.on);
const unsub = bus.subscribe((signal) => {
if (watchedSenses.has(signal.senseId)) {
workflowTriggerFn(workflowName, signal.payload);
}
});
unsubscribers.push(unsub);
}
continue;
}
if (reflex.kind !== "sense") continue;
const senseReflex = reflex;
const senseName = senseReflex.sense;
+15 -4
View File
@@ -1,5 +1,5 @@
import { readFileSync, readdirSync } from "node:fs";
import { join } from "node:path";
import { mkdirSync, readFileSync, readdirSync } from "node:fs";
import { dirname, join } from "node:path";
import Database from "better-sqlite3";
import { drizzle } from "drizzle-orm/better-sqlite3";
@@ -8,6 +8,8 @@ import type { BetterSQLite3Database } from "drizzle-orm/better-sqlite3";
import type { Result } from "@uncaged/nerve-core";
import { err, ok } from "@uncaged/nerve-core";
import type { BlobStore } from "./blob-store.js";
/** A Drizzle DB instance (schema-generic) */
export type DrizzleDB = BetterSQLite3Database<Record<string, never>>;
@@ -17,11 +19,14 @@ export type PeerMap = Readonly<Record<string, DrizzleDB>>;
/** Options passed to a compute function */
export type ComputeOptions = {
signal: AbortSignal;
/** CAS under `data/blobs/`; injected by the sense worker when available. */
blobs?: BlobStore;
};
/**
* The shape every sense's index.ts must export.
* Engine injects `db` (read-write), `peers` (read-only), and `options`.
* Engine injects `db` (read-write), `peers` (read-only), and `options`
* (`signal`, and `blobs` when running in the sense worker — RFC-001 §8 CAS).
* Returns T when a signal should be emitted, null for silence.
*/
export type ComputeFn<T = unknown> = (
@@ -128,6 +133,7 @@ export function openSenseDb(
let sqlite: Database.Database;
try {
mkdirSync(dirname(dbPath), { recursive: true });
sqlite = new Database(dbPath);
// WAL mode for better concurrent read performance
sqlite.pragma("journal_mode = WAL");
@@ -191,14 +197,19 @@ export async function loadComputeFn(senseIndexPath: string): Promise<Result<Comp
* Execute a sense's compute function with an optional soft timeout.
* If timeoutMs is provided and compute takes longer, the AbortSignal is
* triggered and an error Result is returned.
* When `blobStore` is set, it is exposed as `options.blobs` (see RFC-001 §8).
*/
export async function executeCompute(
runtime: SenseRuntime,
peers: PeerMap,
timeoutMs?: number,
blobStore?: BlobStore,
): Promise<Result<unknown | null>> {
const controller = new AbortController();
const options: ComputeOptions = { signal: controller.signal };
const options: ComputeOptions =
blobStore !== undefined
? { signal: controller.signal, blobs: blobStore }
: { signal: controller.signal };
let timer: ReturnType<typeof setTimeout> | undefined;
const timeoutPromise =
+8 -3
View File
@@ -10,6 +10,7 @@
* senses/<name>/index.js ← compiled compute
* senses/<name>/migrations/ ← SQL migration files
* data/senses/<name>.db ← SQLite data file
* data/blobs/<aa>/<hashrest> ← CAS (sha256), via options.blobs in compute
* nerve.yaml ← config
*/
@@ -19,6 +20,7 @@ import { join, resolve } from "node:path";
import { parseNerveConfig } from "@uncaged/nerve-core";
import type { NerveConfig } from "@uncaged/nerve-core";
import { createBlobStore } from "./blob-store.js";
import type { WorkerToParentMessage } from "./ipc.js";
import { parseParentMessage } from "./ipc.js";
import { executeCompute, loadComputeFn, openPeerDb, openSenseDb } from "./sense-runtime.js";
@@ -162,9 +164,10 @@ async function runCompute(
peers: PeerMap,
timeoutMs: number,
gracePeriodMs: number | null,
blobStore: ReturnType<typeof createBlobStore>,
): Promise<void> {
try {
const result = await executeCompute(runtime, peers, timeoutMs);
const result = await executeCompute(runtime, peers, timeoutMs, blobStore);
if (!result.ok) {
sendError(senseName, result.error.message);
if (gracePeriodMs !== null && result.error.message.includes("timed out")) {
@@ -193,6 +196,7 @@ function handleMessage(
group: string,
senseConfigs: Map<string, { timeout: number | null; gracePeriod: number | null }>,
inFlight: Map<string, Promise<void>>,
blobStore: ReturnType<typeof createBlobStore>,
): void {
const parseResult = parseParentMessage(raw);
if (!parseResult.ok) {
@@ -230,7 +234,7 @@ function handleMessage(
const previous = inFlight.get(msg.sense) ?? Promise.resolve();
const next = previous
.then(() => runCompute(msg.sense, runtime, peers, timeoutMs, gracePeriodMs))
.then(() => runCompute(msg.sense, runtime, peers, timeoutMs, gracePeriodMs, blobStore))
.catch((e: unknown) => {
const errMsg = e instanceof Error ? e.message : String(e);
sendError(msg.sense, errMsg);
@@ -294,11 +298,12 @@ async function bootstrap(nerveRoot: string, group: string): Promise<void> {
}
const inFlight = new Map<string, Promise<void>>();
const blobStore = createBlobStore(join(nerveRoot, "data", "blobs"));
sendReady();
process.on("message", (raw: unknown) => {
handleMessage(raw, runtimes, peers, group, senseConfigs, inFlight);
handleMessage(raw, runtimes, peers, group, senseConfigs, inFlight, blobStore);
});
}
+282 -82
View File
@@ -13,9 +13,15 @@ import { fileURLToPath } from "node:url";
import type { NerveConfig, WorkflowConfig } from "@uncaged/nerve-core";
import type { ShutdownMessage, StartThreadMessage, ThreadEventMessage } from "./ipc.js";
import type {
ResumeThreadMessage,
ShutdownMessage,
StartThreadMessage,
ThreadEventMessage,
} from "./ipc.js";
import { parseWorkerMessage } from "./ipc.js";
import type { LogStore } from "./log-store.js";
import type { WorkflowRunStatus } from "./log-store.js";
export type WorkflowManager = {
/** Trigger a new workflow thread (called by Reflex scheduler). */
@@ -24,6 +30,16 @@ export type WorkflowManager = {
activeCount: (workflowName: string) => number;
/** Number of pending queued threads waiting to run for a workflow. */
queueLength: (workflowName: string) => number;
/** Total active workflow threads across all workflows. */
totalActiveCount: () => number;
/** Update the config reference (e.g. after hot reload). Active workers are unaffected. */
updateConfig: (newConfig: NerveConfig) => void;
/**
* Drain active threads for a workflow, then respawn its worker process.
* Used for hot reload when the workflow .ts file changes.
* Waits up to `drainTimeoutMs` for threads to complete before force-killing.
*/
drainAndRespawn: (workflowName: string, drainTimeoutMs?: number) => Promise<void>;
/** Gracefully shut down all workflow workers. */
stop: () => Promise<void>;
};
@@ -42,8 +58,21 @@ type WorkerEntry = {
workflowName: string;
process: ChildProcess;
stopping: boolean;
/** When set, the worker is draining before a hot-reload respawn. */
draining: boolean;
};
// Crash respawn backoff: track crash timestamps per workflow.
const MAX_CRASHES_IN_WINDOW = 5;
const CRASH_WINDOW_MS = 60_000;
/**
* Worker shutdown timeout — must stay in sync with SHUTDOWN_TIMEOUT_MS in workflow-worker.ts.
* The drain timeout passed to drainAndRespawn must be >= this value so the worker has
* enough time to finish in-flight threads before the parent force-kills it.
*/
const WORKER_SHUTDOWN_TIMEOUT_MS = 10_000;
const DEFAULT_MAX_QUEUE = 100;
function resolveWorkerScript(): string {
@@ -57,9 +86,16 @@ function spawnWorkflowWorker(
workflowName: string,
workerScript: string,
): ChildProcess {
return fork(workerScript, ["--workflow", workflowName, "--root", nerveRoot], {
const child = fork(workerScript, ["--workflow", workflowName, "--root", nerveRoot], {
stdio: ["ignore", "inherit", "inherit", "ipc"],
});
// Prevent unhandled EPIPE when writing to a child whose IPC channel closed
child.on("error", (err) => {
if ((err as NodeJS.ErrnoException).code !== "EPIPE") {
console.error("[worker] error:", err.message);
}
});
return child;
}
function sendStartThread(worker: ChildProcess, msg: StartThreadMessage): void {
@@ -82,6 +118,15 @@ function sendShutdown(worker: ChildProcess, entry: WorkerEntry): void {
}
}
function sendResumeThread(worker: ChildProcess, msg: ResumeThreadMessage): void {
if (worker.connected === false) return;
try {
worker.send(msg);
} catch {
// IPC channel closed between connected check and send
}
}
function waitForExit(child: ChildProcess, timeoutMs: number): Promise<void> {
return new Promise((resolve) => {
const timer = setTimeout(() => {
@@ -97,14 +142,16 @@ function waitForExit(child: ChildProcess, timeoutMs: number): Promise<void> {
export function createWorkflowManager(
nerveRoot: string,
config: NerveConfig,
initialConfig: NerveConfig,
logStore: LogStore,
): WorkflowManager {
const workerScript = resolveWorkerScript();
const states = new Map<string, WorkflowState>();
const workers = new Map<string, WorkerEntry>();
const crashTimestamps = new Map<string, number[]>();
let stopped = false;
let config = initialConfig;
function getOrCreateState(workflowName: string): WorkflowState {
let state = states.get(workflowName);
@@ -119,6 +166,19 @@ export function createWorkflowManager(
return config.workflows?.[workflowName] ?? null;
}
function toWorkflowRunStatus(eventType: string): WorkflowRunStatus | null {
const map: Record<string, WorkflowRunStatus> = {
started: "started",
queued: "queued",
completed: "completed",
failed: "failed",
crashed: "crashed",
dropped: "dropped",
interrupted: "interrupted",
};
return map[eventType] ?? null;
}
function logWorkflowEvent(
workflowName: string,
runId: string,
@@ -126,34 +186,12 @@ export function createWorkflowManager(
payload?: unknown,
): void {
const ts = Date.now();
if (
eventType === "started" ||
eventType === "queued" ||
eventType === "completed" ||
eventType === "failed" ||
eventType === "crashed" ||
eventType === "dropped"
) {
const status =
eventType === "dropped"
? ("dropped" as const)
: eventType === "queued"
? ("queued" as const)
: eventType === "started"
? ("started" as const)
: eventType === "completed"
? ("completed" as const)
: eventType === "failed"
? ("failed" as const)
: ("crashed" as const);
const serialised = payload !== undefined ? JSON.stringify(payload) : null;
const status = toWorkflowRunStatus(eventType);
if (status !== null) {
logStore.upsertWorkflowRun(
{
source: "workflow",
type: eventType,
refId: runId,
payload: payload !== undefined ? JSON.stringify(payload) : null,
ts,
},
{ source: "workflow", type: eventType, refId: runId, payload: serialised, ts },
{ runId, workflow: workflowName, status, ts },
);
} else {
@@ -161,7 +199,7 @@ export function createWorkflowManager(
source: "workflow",
type: eventType,
refId: runId,
payload: payload !== undefined ? JSON.stringify(payload) : null,
payload: serialised,
ts,
});
}
@@ -179,7 +217,8 @@ export function createWorkflowManager(
triggerPayload: payload,
};
sendStartThread(worker.process, msg);
logWorkflowEvent(workflowName, runId, "started");
// Store triggerPayload in the log so it can be recovered after a crash
logWorkflowEvent(workflowName, runId, "started", { triggerPayload: payload });
}
function dequeueNext(workflowName: string): void {
@@ -211,6 +250,60 @@ export function createWorkflowManager(
}
}
function recoverQueuedRun(workflowName: string, runId: string, state: WorkflowState): void {
if (state.queue.some((q) => q.runId === runId)) return;
const triggerPayload = logStore.getTriggerPayload(runId);
state.queue.push({ runId, payload: triggerPayload });
process.stderr.write(
`[workflow-manager] crash-recovery: re-queued thread "${runId}" for "${workflowName}"\n`,
);
}
function recoverStartedRun(
workflowName: string,
runId: string,
state: WorkflowState,
worker: WorkerEntry,
): void {
if (state.active.has(runId)) return;
const events = logStore.getThreadEvents(runId);
const triggerPayload = logStore.getTriggerPayload(runId);
state.active.add(runId);
const msg: ResumeThreadMessage = {
type: "resume-thread",
runId,
events,
triggerPayload,
};
sendResumeThread(worker.process, msg);
process.stderr.write(
`[workflow-manager] crash-recovery: resuming thread "${runId}" for "${workflowName}" (${events.length} events)\n`,
);
}
function recoverThreadsForWorker(workflowName: string, worker: WorkerEntry): void {
const activeRuns = logStore.getActiveWorkflowRuns(workflowName);
const state = getOrCreateState(workflowName);
for (const run of activeRuns) {
if (run.status === "queued") {
recoverQueuedRun(workflowName, run.runId, state);
} else if (run.status === "started") {
recoverStartedRun(workflowName, run.runId, state, worker);
}
}
}
function recordCrashAndCheckLimit(workflowName: string): boolean {
const now = Date.now();
const timestamps = (crashTimestamps.get(workflowName) ?? []).filter(
(t) => now - t < CRASH_WINDOW_MS,
);
timestamps.push(now);
crashTimestamps.set(workflowName, timestamps);
return timestamps.length > MAX_CRASHES_IN_WINDOW;
}
function handleWorkerCrash(workflowName: string): void {
const state = states.get(workflowName);
if (state === undefined) return;
@@ -220,13 +313,113 @@ export function createWorkflowManager(
process.stderr.write(
`[workflow-manager] worker for "${workflowName}" crashed with ${crashedCount} active thread(s)\n`,
);
for (const runId of state.active) {
logWorkflowEvent(workflowName, runId, "crashed");
}
}
// All active threads are now crashed — we can't recover runIds from this
// in-memory state alone (Phase 3 crash recovery will use the DB).
// Reset active set so the manager stays consistent.
state.active.clear();
workers.delete(workflowName);
if (stopped || workflowConfig(workflowName) === null) return;
if (recordCrashAndCheckLimit(workflowName)) {
const count = crashTimestamps.get(workflowName)?.length ?? 0;
process.stderr.write(
`[workflow-manager] worker for "${workflowName}" crashed ${count} times in ${CRASH_WINDOW_MS}ms — stopping respawn\n`,
);
return;
}
process.stderr.write(
`[workflow-manager] respawning worker for "${workflowName}" after crash\n`,
);
const newWorker = getOrSpawnWorker(workflowName);
setImmediate(() => {
recoverThreadsForWorker(workflowName, newWorker);
});
}
function handleWorkerMessage(workflowName: string, raw: unknown): void {
const result = parseWorkerMessage(raw);
if (!result.ok) {
process.stderr.write(
`[workflow-manager] invalid message from "${workflowName}" worker: ${result.error.message}\n`,
);
return;
}
const msg = result.value;
if (msg.type === "thread-event") {
handleThreadEvent(workflowName, msg);
return;
}
if (msg.type === "thread-command-event") {
logStore.append({
source: "workflow",
type: "thread_command_event",
refId: msg.runId,
payload: JSON.stringify(msg.event),
ts: Date.now(),
});
return;
}
if (msg.type === "workflow-error") {
process.stderr.write(
`[workflow-manager] workflow-error for runId "${msg.runId}" in "${workflowName}": ${msg.error}\n`,
);
const state = states.get(workflowName);
if (state !== undefined) {
state.active.delete(msg.runId);
dequeueNext(workflowName);
}
logWorkflowEvent(workflowName, msg.runId, "failed", { error: msg.error });
return;
}
if (msg.type === "error") {
process.stderr.write(
`[workflow-manager] error from "${workflowName}" worker: ${msg.error}\n`,
);
}
}
function markActiveRunsInterrupted(workflowName: string): void {
const state = states.get(workflowName);
if (state === undefined) return;
for (const runId of state.active) {
logWorkflowEvent(workflowName, runId, "interrupted");
}
state.active.clear();
}
function handleWorkerExit(workflowName: string, code: number | null): void {
const entry = workers.get(workflowName);
if (entry?.draining) {
workers.delete(workflowName);
markActiveRunsInterrupted(workflowName);
if (!stopped && workflowConfig(workflowName) !== null) {
process.stderr.write(
`[workflow-manager] worker for "${workflowName}" drained, respawning\n`,
);
getOrSpawnWorker(workflowName);
}
return;
}
if (entry?.stopping) {
workers.delete(workflowName);
const state = states.get(workflowName);
if (state !== undefined) {
state.active.clear();
}
return;
}
process.stderr.write(
`[workflow-manager] worker for "${workflowName}" exited with code ${code ?? "null"}\n`,
);
handleWorkerCrash(workflowName);
}
function getOrSpawnWorker(workflowName: string): WorkerEntry {
@@ -238,57 +431,14 @@ export function createWorkflowManager(
const child = spawnWorkflowWorker(nerveRoot, workflowName, workerScript);
child.on("message", (raw: unknown) => {
const result = parseWorkerMessage(raw);
if (!result.ok) {
process.stderr.write(
`[workflow-manager] invalid message from "${workflowName}" worker: ${result.error.message}\n`,
);
return;
}
const msg = result.value;
if (msg.type === "thread-event") {
handleThreadEvent(workflowName, msg);
return;
}
if (msg.type === "workflow-error") {
process.stderr.write(
`[workflow-manager] workflow-error for runId "${msg.runId}" in "${workflowName}": ${msg.error}\n`,
);
const state = states.get(workflowName);
if (state !== undefined) {
state.active.delete(msg.runId);
dequeueNext(workflowName);
}
logWorkflowEvent(workflowName, msg.runId, "failed", { error: msg.error });
return;
}
if (msg.type === "error") {
process.stderr.write(
`[workflow-manager] error from "${workflowName}" worker: ${msg.error}\n`,
);
}
handleWorkerMessage(workflowName, raw);
});
child.on("exit", (code) => {
const entry = workers.get(workflowName);
if (entry?.stopping) {
workers.delete(workflowName);
const state = states.get(workflowName);
if (state !== undefined) {
state.active.clear();
}
return;
}
process.stderr.write(
`[workflow-manager] worker for "${workflowName}" exited with code ${code ?? "null"}\n`,
);
handleWorkerCrash(workflowName);
handleWorkerExit(workflowName, code);
});
const entry: WorkerEntry = { workflowName, process: child, stopping: false };
const entry: WorkerEntry = { workflowName, process: child, stopping: false, draining: false };
workers.set(workflowName, entry);
return entry;
}
@@ -348,6 +498,48 @@ export function createWorkflowManager(
return states.get(workflowName)?.queue.length ?? 0;
}
function totalActiveCount(): number {
let total = 0;
for (const state of states.values()) {
total += state.active.size;
}
return total;
}
function updateConfig(newConfig: NerveConfig): void {
config = newConfig;
}
/**
* Default drain timeout must be at least WORKER_SHUTDOWN_TIMEOUT_MS so the worker
* has enough time to finish in-flight threads before the parent force-kills it.
*/
const DEFAULT_DRAIN_TIMEOUT_MS = Math.max(30_000, WORKER_SHUTDOWN_TIMEOUT_MS + 5_000);
async function drainAndRespawn(
workflowName: string,
drainTimeoutMs: number = DEFAULT_DRAIN_TIMEOUT_MS,
): Promise<void> {
const entry = workers.get(workflowName);
if (entry === undefined) {
// No active worker — nothing to drain
return;
}
entry.draining = true;
// Send shutdown without setting stopping=true (so the exit handler uses the draining branch)
if (entry.process.connected) {
const msg: ShutdownMessage = { type: "shutdown" };
try {
entry.process.send(msg);
} catch {
// IPC closed
}
}
await waitForExit(entry.process, drainTimeoutMs);
// The exit handler (draining branch) will respawn the worker automatically
}
async function stop(): Promise<void> {
stopped = true;
const exitPromises: Promise<void>[] = [];
@@ -359,5 +551,13 @@ export function createWorkflowManager(
workers.clear();
}
return { startWorkflow, activeCount, queueLength, stop };
return {
startWorkflow,
activeCount,
queueLength,
totalActiveCount,
updateConfig,
drainAndRespawn,
stop,
};
}
+106 -9
View File
@@ -9,8 +9,8 @@
* workflows/<name>/index.ts (or .js) ← user workflow definition
*/
import { resolve, join } from "node:path";
import { existsSync } from "node:fs";
import { join, resolve } from "node:path";
import type {
CommandEvent,
@@ -19,7 +19,7 @@ import type {
WorkflowDefinition,
} from "@uncaged/nerve-core";
import type { WorkerToParentMessage, ThreadEventType } from "./ipc.js";
import type { ThreadCommandEventMessage, ThreadEventType, WorkerToParentMessage } from "./ipc.js";
import { parseParentMessage } from "./ipc.js";
// ---------------------------------------------------------------------------
@@ -44,29 +44,107 @@ function sendWorkflowError(runId: string, error: string): void {
send({ type: "workflow-error", runId, error });
}
function sendCommandEvent(runId: string, event: CommandEvent): void {
const msg: ThreadCommandEventMessage = {
type: "thread-command-event",
runId,
event: event as { type: string; [key: string]: unknown },
};
send(msg);
}
// ---------------------------------------------------------------------------
// Thread loop (RFC-002 §5.4)
// ---------------------------------------------------------------------------
/**
* Replay persisted events through moderate() to reconstruct ThreadState,
* then execute the next role and return the resulting CommandEvent.
* Returns null if the thread is already complete (moderate returned null).
*/
async function replayAndResume(
def: WorkflowDefinition,
runId: string,
ctx: WorkflowContext,
state: ThreadState,
resumeEvents: CommandEvent[],
): Promise<CommandEvent | null> {
let lastNext: ReturnType<typeof def.moderate> = null;
for (const ev of resumeEvents) {
state.events.push(ev);
lastNext = def.moderate(state, ev);
if (lastNext === null) {
sendThreadEvent(runId, "completed", null);
return null;
}
}
const next = lastNext;
if (next === null) {
sendThreadEvent(runId, "completed", null);
return null;
}
const role = def.roles[next.role];
if (!role) {
sendWorkflowError(runId, `Unknown role: ${next.role}`);
return null;
}
try {
const event = await role.execute(next.prompt, ctx);
sendCommandEvent(runId, event);
return event;
} catch (e: unknown) {
const errMsg = e instanceof Error ? e.message : String(e);
sendThreadEvent(runId, "failed", { error: errMsg });
return null;
}
}
async function runThread(
def: WorkflowDefinition,
workflowName: string,
runId: string,
triggerPayload: unknown,
/** Pre-existing event history for crash-recovery resume. Empty for a fresh thread. */
resumeEvents: CommandEvent[] = [],
): Promise<void> {
const state: ThreadState = { runId, events: [] };
const ctx: WorkflowContext = {
runId,
workflowName,
log: (msg) =>
sendThreadEvent(runId, "step_complete", { message: msg }),
log: (msg) => sendThreadEvent(runId, "step_complete", { message: msg }),
};
let event: CommandEvent = {
const initialEvent: CommandEvent = {
type: "thread_start",
triggerPayload: triggerPayload != null && typeof triggerPayload === "object" ? triggerPayload : {},
triggerPayload:
triggerPayload != null && typeof triggerPayload === "object" ? triggerPayload : {},
};
// On resume: replay persisted events, run the next un-executed role, then continue.
if (resumeEvents.length > 0) {
const nextEvent = await replayAndResume(def, runId, ctx, state, resumeEvents);
if (nextEvent === null) return;
await continueThread(def, runId, ctx, state, nextEvent);
return;
}
// Fresh thread — send the initial command event and enter the loop.
sendCommandEvent(runId, initialEvent);
await continueThread(def, runId, ctx, state, initialEvent);
}
async function continueThread(
def: WorkflowDefinition,
runId: string,
ctx: WorkflowContext,
state: ThreadState,
firstEvent: CommandEvent,
): Promise<void> {
let event = firstEvent;
const MAX_STEPS = 1000;
let step = 0;
while (step < MAX_STEPS) {
@@ -92,6 +170,7 @@ async function runThread(
sendThreadEvent(runId, "failed", { error: errMsg });
return;
}
sendCommandEvent(runId, event);
}
if (step >= MAX_STEPS) {
sendWorkflowError(runId, `Thread exceeded maximum steps (${MAX_STEPS})`);
@@ -114,8 +193,7 @@ async function loadWorkflowDefinition(
const indexPath = candidates.find((p) => existsSync(p));
if (!indexPath) {
throw new Error(
`Workflow definition not found for "${workflowName}". Tried:\n` +
candidates.map((p) => ` ${p}`).join("\n"),
`Workflow definition not found for "${workflowName}". Tried:\n${candidates.map((p) => ` ${p}`).join("\n")}`,
);
}
@@ -157,7 +235,7 @@ function handleMessage(
if (msg.type === "shutdown") {
shuttingDown.value = true;
const timeout = new Promise<void>(r => setTimeout(r, 10_000));
const timeout = new Promise<void>((r) => setTimeout(r, 10_000));
Promise.race([Promise.all(inFlight.values()), timeout])
.then(() => process.exit(0))
.catch(() => process.exit(1));
@@ -182,6 +260,25 @@ function handleMessage(
inFlight.set(runId, next);
return;
}
if (msg.type === "resume-thread") {
if (shuttingDown.value) return;
const { runId, events, triggerPayload } = msg;
const previous = inFlight.get(runId) ?? Promise.resolve();
const next = previous
.then(() => runThread(def, workflowName, runId, triggerPayload, events as CommandEvent[]))
.catch((e: unknown) => {
const errMsg = e instanceof Error ? e.message : String(e);
sendWorkflowError(runId, errMsg);
})
.finally(() => {
inFlight.delete(runId);
});
inFlight.set(runId, next);
return;
}
}
// ---------------------------------------------------------------------------
+57 -3
View File
@@ -23,16 +23,22 @@ importers:
'@uncaged/nerve-core':
specifier: workspace:*
version: link:../core
'@uncaged/nerve-daemon':
specifier: workspace:*
version: link:../daemon
citty:
specifier: ^0.1.6
version: 0.1.6
devDependencies:
'@types/better-sqlite3':
specifier: ^7.6.13
version: 7.6.13
'@types/node':
specifier: ^22.0.0
version: 22.19.17
'@uncaged/nerve-daemon':
specifier: workspace:*
version: link:../daemon
vitest:
specifier: ^4.1.5
version: 4.1.5(@types/node@22.19.17)(vite@8.0.9(@types/node@22.19.17)(esbuild@0.27.7)(yaml@2.8.3))
packages/core:
dependencies:
@@ -1579,6 +1585,14 @@ snapshots:
chai: 6.2.2
tinyrainbow: 3.1.0
'@vitest/mocker@4.1.5(vite@8.0.9(@types/node@22.19.17)(esbuild@0.27.7)(yaml@2.8.3))':
dependencies:
'@vitest/spy': 4.1.5
estree-walker: 3.0.3
magic-string: 0.30.21
optionalDependencies:
vite: 8.0.9(@types/node@22.19.17)(esbuild@0.27.7)(yaml@2.8.3)
'@vitest/mocker@4.1.5(vite@8.0.9(@types/node@25.6.0)(esbuild@0.27.7)(yaml@2.8.3))':
dependencies:
'@vitest/spy': 4.1.5
@@ -2089,6 +2103,19 @@ snapshots:
util-deprecate@1.0.2: {}
vite@8.0.9(@types/node@22.19.17)(esbuild@0.27.7)(yaml@2.8.3):
dependencies:
lightningcss: 1.32.0
picomatch: 4.0.4
postcss: 8.5.10
rolldown: 1.0.0-rc.16
tinyglobby: 0.2.16
optionalDependencies:
'@types/node': 22.19.17
esbuild: 0.27.7
fsevents: 2.3.3
yaml: 2.8.3
vite@8.0.9(@types/node@25.6.0)(esbuild@0.27.7)(yaml@2.8.3):
dependencies:
lightningcss: 1.32.0
@@ -2102,6 +2129,33 @@ snapshots:
fsevents: 2.3.3
yaml: 2.8.3
vitest@4.1.5(@types/node@22.19.17)(vite@8.0.9(@types/node@22.19.17)(esbuild@0.27.7)(yaml@2.8.3)):
dependencies:
'@vitest/expect': 4.1.5
'@vitest/mocker': 4.1.5(vite@8.0.9(@types/node@22.19.17)(esbuild@0.27.7)(yaml@2.8.3))
'@vitest/pretty-format': 4.1.5
'@vitest/runner': 4.1.5
'@vitest/snapshot': 4.1.5
'@vitest/spy': 4.1.5
'@vitest/utils': 4.1.5
es-module-lexer: 2.0.0
expect-type: 1.3.0
magic-string: 0.30.21
obug: 2.1.1
pathe: 2.0.3
picomatch: 4.0.4
std-env: 4.1.0
tinybench: 2.9.0
tinyexec: 1.1.1
tinyglobby: 0.2.16
tinyrainbow: 3.1.0
vite: 8.0.9(@types/node@22.19.17)(esbuild@0.27.7)(yaml@2.8.3)
why-is-node-running: 2.3.0
optionalDependencies:
'@types/node': 22.19.17
transitivePeerDependencies:
- msw
vitest@4.1.5(@types/node@25.6.0)(vite@8.0.9(@types/node@25.6.0)(esbuild@0.27.7)(yaml@2.8.3)):
dependencies:
'@vitest/expect': 4.1.5
+8
View File
@@ -0,0 +1,8 @@
#!/bin/bash
# All packages must use pnpm publish. Block npm publish unconditionally.
if [ -z "$npm_execpath" ] || [[ "$npm_execpath" != *pnpm* ]]; then
echo "❌ Use 'pnpm publish' instead of 'npm publish'."
echo " pnpm auto-converts workspace:* dependencies to real versions."
exit 1
fi