Repository navigation
feat(cron): add scheduled task execution system with AI-callable tools - #872
Closed
kanakagrawal-crypto wants to merge 1 commit into
Closed
kanakagrawal-crypto wants to merge 1 commit into
kanakagrawal-crypto wants to merge 1 commit into
Conversation
|
@claude is attempting to deploy a commit to the Sachin Sharma's projects Team on Vercel. A member of the Team first needs to authorize it. |
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughAdds a full cron/scheduling subsystem: types, schedulers (BullMQ + NodeTimeout), task stores (in-memory, Redis), CronManager orchestration, AI-facing cron tools, CLI/SDK integration, and tooling/utility changes to expose and gate cron tools. Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant CLI
participant CronTools
participant CronMgr
participant Scheduler
participant Store
participant Executor
User->>CLI: invoke create scheduled task
CLI->>CronTools: createScheduledTask(prompt, schedule)
CronTools->>CronMgr: createTask(options)
CronMgr->>Store: save(task)
Store-->>CronMgr: persisted
CronMgr->>Scheduler: schedule(task, callback)
Scheduler-->>CronMgr: scheduled
Note over Scheduler: waits until next run
Scheduler->>CronMgr: trigger executeTask(taskId)
CronMgr->>Executor: execute(task, sessionId)
Executor-->>CronMgr: result (responseText, tokens)
CronMgr->>Store: addRunResult(taskId, run)
CronMgr->>Scheduler: getNextRunTime(task)
Scheduler-->>CronMgr: nextRunAt
CronMgr->>Store: update task nextRunAt/status
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 16
🧹 Nitpick comments (2)
src/lib/core/modules/ToolsManager.ts (1)
220-234: Consider caching getter result to avoid repeated evaluations.The
directToolsGetter()is called multiple times within this method (lines 222, 223, 229, 232-233). Each call re-evaluatesshouldDisableBuiltinTools()and spreadsdirectAgentToolswithgetCronTools()in the BaseProvider getter.Consider caching the result at the start of the method:
♻️ Suggested optimization
private async processDirectTools(tools: Record<string, Tool>): Promise<void> { + const directTools = this.directToolsGetter(); if ( - !this.directToolsGetter() || - Object.keys(this.directToolsGetter()).length === 0 + !directTools || + Object.keys(directTools).length === 0 ) { return; } logger.debug( - `[ToolsManager] Loading ${Object.keys(this.directToolsGetter()).length} direct tools`, + `[ToolsManager] Loading ${Object.keys(directTools).length} direct tools`, ); - for (const [toolName, directTool] of Object.entries( - this.directToolsGetter(), - )) { + for (const [toolName, directTool] of Object.entries(directTools)) {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/core/modules/ToolsManager.ts` around lines 220 - 234, The method processDirectTools repeatedly calls this.directToolsGetter(), causing redundant evaluations; cache the getter result once at the start (e.g., const directTools = this.directToolsGetter()) and use that cached variable for the empty-check, for logger message (Object.keys(directTools).length) and for the for-loop (Object.entries(directTools)), ensuring all previous calls to this.directToolsGetter() in processDirectTools are replaced with the single cached reference to avoid re-evaluating shouldDisableBuiltinTools()/getCronTools().src/cli/factories/commandFactory.ts (1)
1958-1964: Cron keep-alive log text is overly strict about eventual completion.
hasActiveCronTasks()checks for"active"tasks, so recurring/future schedules can keep the process alive long-term. The current message (“until tasks complete”) can be misleading for recurring jobs. Consider wording that reflects “while active schedules exist.”🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/cli/factories/commandFactory.ts` around lines 1958 - 1964, The logger message after checking globalSession.getCurrentSessionId() and sdk.hasActiveCronTasks() is misleading about eventual completion; update the logger.info call (the message emitted when sdk.hasActiveCronTasks() is true) to reflect that the process will remain alive while active scheduled tasks or recurring schedules exist (e.g., “Process will stay alive while active scheduled tasks exist; this may include recurring schedules.”) so it correctly describes behavior of hasActiveCronTasks().
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/cli/factories/commandFactory.ts`:
- Around line 2926-2933: The cron liveness check is creating a fresh SDK via
globalSession.getOrCreateNeuroLink() (streamSdk) which can differ from the SDK
used in executeRealStream, so hasActiveCronTasks() may miss tasks; change the
code to reuse the same NeuroLink/SDK instance used for streaming (the instance
referenced/created for executeRealStream) when calling hasActiveCronTasks(), or
thread that same instance into this scope (rather than calling
getOrCreateNeuroLink() again), so the liveness check observes the same scheduled
tasks created by executeRealStream.
In `@src/lib/agent/directTools.ts`:
- Line 806: directAgentTools currently spreads createCronTools(() =>
cronManagerRef) at module build time which bypasses shouldDisableCronTools() and
makes cron tools always available; remove the unconditional spread from
directAgentTools and instead expose cron tools only via getCronTools() (which
should call shouldDisableCronTools() internally) and update any consumers that
iterate directAgentTools (e.g., MCP registry/server) to call getCronTools() and
merge its returned tools at runtime; ensure createCronTools, cronManagerRef,
getCronTools, and shouldDisableCronTools are used as the single source of truth
for cron tool registration so disable controls are respected.
In `@src/lib/cron/cronManager.ts`:
- Around line 85-109: When creating (and updating) scheduled tasks, add a guard
so sessionMode === "same-session" must have a non-empty creatorSessionId; in
createTask (where ScheduledTask is built: id/generateTaskId, sessionMode,
creatorSessionId) throw an error if options.sessionMode === "same-session" &&
!options.creatorSessionId to prevent silently falling back to the
`cron:${taskId}` session; apply the same validation in the corresponding
updateTask path that mutates task.sessionMode or creatorSessionId to ensure you
never accept "same-session" without a creatorSessionId.
- Around line 233-240: Wrap the call to this.executor(task, sessionId) in the
withTimeout(...) helper so provider calls cannot hang indefinitely; catch a
timeout rejection and convert it into a failed TaskRunResult (set run.status =
"failed", run.completedAt, run.durationMs, and populate
run.responseText/tokenUsage or an error message) before continuing. Update the
block that currently assigns result.responseText and result.tokenUsage to
instead handle either a normal result or the timeout/error result, ensuring the
rest of the TaskRun state updates (run.status, run.completedAt, run.durationMs)
happen in both success and timeout/error paths.
- Around line 57-62: The code currently silently falls back to InMemoryTaskStore
when this.config.store === "redis" but this.config.redisConfig is missing;
change this in the CronManager initialization so you do not silently downgrade:
either validate and throw a configuration error when this.config.store ===
"redis" && !this.config.redisConfig (e.g., throw new Error('Redis selected as
store but redisConfig is missing')) or construct RedisTaskStore with sensible
defaults if you prefer a fallback; update the branch that now instantiates
InMemoryTaskStore to instead perform this validation and only create
InMemoryTaskStore when this.config.store !== "redis" (referencing
this.config.store, this.config.redisConfig, RedisTaskStore and
InMemoryTaskStore).
In `@src/lib/cron/schedulerBackend.ts`:
- Around line 147-149: The interval runner created in scheduleInterval (the
"every" schedule) uses setInterval which fires regardless of the async callback
state, allowing concurrent runs; change it to a protected self-rescheduling loop
(use setTimeout) or add a per-task boolean lock to prevent overlap: when
scheduleInterval starts, set a running flag (or use the same protect logic as
scheduleCron) and on each tick check the flag, skip or delay if already running,
otherwise set running=true, await callback(), finally set running=false and then
schedule the next setTimeout; update any timer handle logic (the timer variable)
so clear()/cancel still works and mirror scheduleCron's protect behavior to
avoid concurrent executions.
- Around line 121-128: The scheduler invokes async callbacks using the `void
callback()` pattern (seen in `schedule()` where timers are created and cleared
via `this.timers` and `setTimeout`) which can produce unhandled promise
rejections; update every `void callback()` call (including the occurrences
inside the immediate-return branch and inside `setTimeout` handlers) to call
`callback().catch(...)` and handle errors (log via the scheduler logger or at
least swallow the error) so rejections are captured while keeping `schedule()`
synchronous; ensure all similar sites (the other occurrences around
`executeTask` invocations) are updated consistently.
- Around line 112-128: The schedule code must validate parsed times and
clamp/split delays to avoid NaN or Node's 32-bit timer overflow: after computing
targetTime (in scheduleAt) check isNaN(targetTime) and treat it as an invalid
schedule (reject, log error, or skip task) instead of passing NaN to setTimeout;
compute delayMs and if isNaN(delayMs) or delayMs <= 0 handle immediately or
reject accordingly; define a MAX_DELAY constant (2147483647) and if delayMs >
MAX_DELAY implement a safe scheduling strategy (either split the wait by
scheduling a setTimeout for MAX_DELAY that re-calculates remaining delay and
continues, or schedule successive MAX_DELAY chunks) so long delays like "25d" do
not coerce to ~1ms; apply the same validation and MAX_DELAY-splitting logic at
the other timer creation sites referenced (the other setTimeout/setInterval
usages in this file) to ensure no NaN or overflow reaches Node timers.
In `@src/lib/cron/taskStore.ts`:
- Around line 168-187: The current get→mutate→save pattern in updateStatus and
addRunResult can overwrite concurrent changes; modify these to perform atomic
updates: either implement optimistic locking using Redis WATCH/MULTI/EXEC around
get + save inside updateStatus and addRunResult (retry on EXEC failure), or
split task storage into independent keys (e.g., task:{id}:meta HSET for
status/updatedAt and task:{id}:runs list with LPUSH/LTRIM for runs) and update
each sub-key with single atomic Redis commands; update references to get/save in
the TaskStore so updateStatus and addRunResult use the chosen atomic path and
preserve runCount/run history correctly.
- Around line 84-113: The getClient() implementation on RedisTaskStore assigns
this.redisClient before the underlying connection is established, which allows
race conditions; change it to memoize the in-flight connect promise (e.g., a
private this.connectPromise or this.redisClientPromise) so concurrent callers
await the same promise, call createClient() but do NOT assign to
this.redisClient until after await client.connect() completes, then set
this.redisClient to the connected client and clear the promise on error or
success; update getClient() to return the already-connected this.redisClient or
await the in-flight promise when present.
In `@src/lib/cron/types.ts`:
- Around line 14-235: The exported cron types (ScheduleType, SessionMode,
TaskStatus, RunStatus, Schedule, TaskRunResult, ScheduledTask,
CreateTaskOptions, TaskFilter, SchedulerBackend, TaskStore, CronManagerConfig,
TaskExecutor) must be moved out of the cron module into the shared types domain
(src/lib/types/) as a new domain file; create the new types file, copy these
exact exported type definitions there, update imports in cron and other modules
to import from the new shared types file, and add a re-export from the original
cron module only if compatibility is needed so callers keep working while you
update references.
In `@src/lib/neurolink.ts`:
- Around line 1398-1399: The global cron manager reference set by
setCronManagerRef(this.cronManager) must be cleared during lifecycle teardown to
avoid use-after-shutdown; add a clearCronManagerRef (or a nullable setter) in
src/lib/agent/directTools.ts that nulls the shared ref and then call that
function from both shutdown() and dispose() in src/lib/neurolink.ts (in the same
locations where setCronManagerRef is used, e.g., the block around
setCronManagerRef(this.cronManager) and the shutdown/dispose implementations) so
the shared pointer is cleared whenever the local manager is closed.
- Around line 1370-1405: Wrap CronManager construction and executor wiring in a
safe guarded block so failures don't throw from the NeuroLink constructor:
attempt to create new CronManager(config) and call cronManager.setExecutor(...)
inside a try/catch (use the withTimeout utility for any async steps, e.g., when
wiring executor calling this.generate), and on any error log it and skip cron
setup (do not call setCronManagerRef or rely on cronManager afterwards); ensure
the code leaves NeuroLink usable even if cron setup fails and that successful
creation still sets the ref and executor on the CronManager instance.
- Around line 2278-2283: Update hasActiveCronTasks to guard against exceptions
and timeouts from cronManager.listTasks: call this.cronManager.listTasks(...)
wrapped with the withTimeout utility and catch any errors; on timeout or
exception log a warning via the existing logger (e.g., processLogger or
this.logger) that includes the caught error and return false so liveness checks
degrade gracefully; keep the original success path (return tasks.length > 0)
unchanged and reference hasActiveCronTasks and this.cronManager.listTasks to
locate the change.
- Around line 1354-1362: The env-derived values for cron config must be
validated before assignment: check envStore (process.env.NEUROLINK_CRON_STORE)
only accept "memory" or "redis" and otherwise fall back to cronConfig?.store or
"memory"; for envMaxConcurrent parseInt(envMaxConcurrent,10) then validate
Number.isInteger and >0 (or at least >=0) and only assign to
CronManagerConfig.maxConcurrentRuns if valid, otherwise fall back to
cronConfig?.maxConcurrentRuns or a sensible default; update the code paths that
build the CronManagerConfig object (references: envStore, envMaxConcurrent,
CronManagerConfig) to perform these guards.
In `@src/lib/session/globalSessionState.ts`:
- Around line 148-155: The cronConfig construction currently only runs when
options.defaultProvider is set so options.defaultModel is ignored; change the
guard to check for either options.defaultProvider or options.defaultModel (e.g.,
if (options?.defaultProvider || options?.defaultModel)) and build the cron
object including defaultProvider and defaultModel only when they exist (use the
existing cronConfig variable and include the fields conditionally) so a call
that supplies defaultModel alone is respected.
---
Nitpick comments:
In `@src/cli/factories/commandFactory.ts`:
- Around line 1958-1964: The logger message after checking
globalSession.getCurrentSessionId() and sdk.hasActiveCronTasks() is misleading
about eventual completion; update the logger.info call (the message emitted when
sdk.hasActiveCronTasks() is true) to reflect that the process will remain alive
while active scheduled tasks or recurring schedules exist (e.g., “Process will
stay alive while active scheduled tasks exist; this may include recurring
schedules.”) so it correctly describes behavior of hasActiveCronTasks().
In `@src/lib/core/modules/ToolsManager.ts`:
- Around line 220-234: The method processDirectTools repeatedly calls
this.directToolsGetter(), causing redundant evaluations; cache the getter result
once at the start (e.g., const directTools = this.directToolsGetter()) and use
that cached variable for the empty-check, for logger message
(Object.keys(directTools).length) and for the for-loop
(Object.entries(directTools)), ensuring all previous calls to
this.directToolsGetter() in processDirectTools are replaced with the single
cached reference to avoid re-evaluating
shouldDisableBuiltinTools()/getCronTools().
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3a5c6b50-b093-4d18-9eda-2de4ef22dddf
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (18)
package.jsonsrc/cli/factories/commandFactory.tssrc/lib/agent/directTools.tssrc/lib/core/baseProvider.tssrc/lib/core/modules/ToolsManager.tssrc/lib/cron/cronManager.tssrc/lib/cron/cronTools.tssrc/lib/cron/index.tssrc/lib/cron/schedulerBackend.tssrc/lib/cron/taskStore.tssrc/lib/cron/types.tssrc/lib/neurolink.tssrc/lib/providers/googleAiStudio.tssrc/lib/providers/googleVertex.tssrc/lib/session/globalSessionState.tssrc/lib/types/configTypes.tssrc/lib/types/index.tssrc/lib/utils/toolUtils.ts
Comment on lines
+233
to
+240
| try { | ||
| const result = await this.executor(task, sessionId); | ||
|
|
||
| run.status = "completed"; | ||
| run.completedAt = Date.now(); | ||
| run.durationMs = run.completedAt - run.startedAt; | ||
| run.responseText = result.responseText; | ||
| run.tokenUsage = result.tokenUsage; |
There was a problem hiding this comment.
Bound scheduled executions with withTimeout(...).
Line 234 waits on provider execution indefinitely. One hung model call can hold a worker slot forever and starve later runs. Wrap the executor call with the repo timeout helper and turn timeout failures into a failed TaskRunResult.
As per coding guidelines: "Use withTimeout utility for async operations and implement graceful degradation with provider fallback for error handling."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib/cron/cronManager.ts` around lines 233 - 240, Wrap the call to
this.executor(task, sessionId) in the withTimeout(...) helper so provider calls
cannot hang indefinitely; catch a timeout rejection and convert it into a failed
TaskRunResult (set run.status = "failed", run.completedAt, run.durationMs, and
populate run.responseText/tokenUsage or an error message) before continuing.
Update the block that currently assigns result.responseText and
result.tokenUsage to instead handle either a normal result or the timeout/error
result, ensuring the rest of the TaskRun state updates (run.status,
run.completedAt, run.durationMs) happen in both success and timeout/error paths.
Comment on lines
+1354
to
+1362
| const envStore = process.env.NEUROLINK_CRON_STORE; | ||
| const envMaxConcurrent = process.env.NEUROLINK_CRON_MAX_CONCURRENT; | ||
|
|
||
| const config: CronManagerConfig = { | ||
| enabled: true, | ||
| store: (envStore as "memory" | "redis") || cronConfig?.store || "memory", | ||
| maxConcurrentRuns: envMaxConcurrent | ||
| ? parseInt(envMaxConcurrent, 10) | ||
| : cronConfig?.maxConcurrentRuns, |
There was a problem hiding this comment.
Validate env-derived cron config values before assigning them.
Line 1359 trusts NEUROLINK_CRON_STORE via type-cast, and Line 1361 can produce NaN for maxConcurrentRuns. Invalid env values can leak into CronManagerConfig and cause startup/runtime instability.
Proposed hardening
- const envStore = process.env.NEUROLINK_CRON_STORE;
- const envMaxConcurrent = process.env.NEUROLINK_CRON_MAX_CONCURRENT;
+ const envStoreRaw = process.env.NEUROLINK_CRON_STORE;
+ const envMaxConcurrentRaw = process.env.NEUROLINK_CRON_MAX_CONCURRENT;
+
+ const envStore =
+ envStoreRaw === "memory" || envStoreRaw === "redis"
+ ? envStoreRaw
+ : undefined;
+
+ const envMaxConcurrent = (() => {
+ if (!envMaxConcurrentRaw) return undefined;
+ const parsed = Number.parseInt(envMaxConcurrentRaw, 10);
+ return Number.isFinite(parsed) && parsed > 0 ? parsed : undefined;
+ })();
@@
- store: (envStore as "memory" | "redis") || cronConfig?.store || "memory",
- maxConcurrentRuns: envMaxConcurrent
- ? parseInt(envMaxConcurrent, 10)
- : cronConfig?.maxConcurrentRuns,
+ store: envStore ?? cronConfig?.store ?? "memory",
+ maxConcurrentRuns: envMaxConcurrent ?? cronConfig?.maxConcurrentRuns,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib/neurolink.ts` around lines 1354 - 1362, The env-derived values for
cron config must be validated before assignment: check envStore
(process.env.NEUROLINK_CRON_STORE) only accept "memory" or "redis" and otherwise
fall back to cronConfig?.store or "memory"; for envMaxConcurrent
parseInt(envMaxConcurrent,10) then validate Number.isInteger and >0 (or at least
>=0) and only assign to CronManagerConfig.maxConcurrentRuns if valid, otherwise
fall back to cronConfig?.maxConcurrentRuns or a sensible default; update the
code paths that build the CronManagerConfig object (references: envStore,
envMaxConcurrent, CronManagerConfig) to perform these guards.
Comment on lines
+1370
to
+1405
| this.cronManager = new CronManager(config); | ||
|
|
||
| // Wire up the executor: when a task fires, call this.generate() | ||
| this.cronManager.setExecutor(async (task, sessionId) => { | ||
| const generateOptions = { | ||
| input: { text: task.prompt }, | ||
| provider: task.provider || config.defaultProvider, | ||
| model: task.model || config.defaultModel, | ||
| sessionId, | ||
| tools: getToolsForCategory("all"), // Enable all tools (file operations, etc.) for cron tasks | ||
| } as Record<string, unknown>; | ||
|
|
||
| const result = await this.generate( | ||
| generateOptions as import("./types/generateTypes.js").GenerateOptions, | ||
| ); | ||
|
|
||
| return { | ||
| responseText: result.content, | ||
| tokenUsage: result.usage | ||
| ? { | ||
| promptTokens: result.usage.input, | ||
| completionTokens: result.usage.output, | ||
| totalTokens: result.usage.total, | ||
| } | ||
| : undefined, | ||
| }; | ||
| }); | ||
|
|
||
| // Set the reference so cron tools can access the manager | ||
| setCronManagerRef(this.cronManager); | ||
|
|
||
| logger.debug("[NeuroLink] CronManager initialized", { | ||
| store: config.store, | ||
| maxConcurrent: config.maxConcurrentRuns, | ||
| }); | ||
| } |
There was a problem hiding this comment.
Cron init failures should not fail NeuroLink construction.
new CronManager(config) and executor wiring are currently unguarded. If cron backend setup fails, constructor initialization fails for the entire SDK instance, even when callers don’t use cron features.
Proposed graceful degradation
- this.cronManager = new CronManager(config);
-
- // Wire up the executor: when a task fires, call this.generate()
- this.cronManager.setExecutor(async (task, sessionId) => {
+ try {
+ this.cronManager = new CronManager(config);
+
+ // Wire up the executor: when a task fires, call this.generate()
+ this.cronManager.setExecutor(async (task, sessionId) => {
const generateOptions = {
input: { text: task.prompt },
provider: task.provider || config.defaultProvider,
model: task.model || config.defaultModel,
sessionId,
tools: getToolsForCategory("all"), // Enable all tools (file operations, etc.) for cron tasks
} as Record<string, unknown>;
const result = await this.generate(
generateOptions as import("./types/generateTypes.js").GenerateOptions,
);
return {
responseText: result.content,
tokenUsage: result.usage
? {
promptTokens: result.usage.input,
completionTokens: result.usage.output,
totalTokens: result.usage.total,
}
: undefined,
};
- });
-
- // Set the reference so cron tools can access the manager
- setCronManagerRef(this.cronManager);
-
- logger.debug("[NeuroLink] CronManager initialized", {
- store: config.store,
- maxConcurrent: config.maxConcurrentRuns,
- });
+ });
+
+ // Set the reference so cron tools can access the manager
+ setCronManagerRef(this.cronManager);
+
+ logger.debug("[NeuroLink] CronManager initialized", {
+ store: config.store,
+ maxConcurrent: config.maxConcurrentRuns,
+ });
+ } catch (error) {
+ this.cronManager = undefined;
+ logger.warn("[NeuroLink] CronManager initialization failed; continuing without cron", {
+ error: error instanceof Error ? error.message : String(error),
+ });
+ }As per coding guidelines, "Use withTimeout utility for async operations and implement graceful degradation with provider fallback for error handling." and "Maintain backward compatibility with existing SDK APIs when making changes. All SDK modifications must not break existing code using the library."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib/neurolink.ts` around lines 1370 - 1405, Wrap CronManager construction
and executor wiring in a safe guarded block so failures don't throw from the
NeuroLink constructor: attempt to create new CronManager(config) and call
cronManager.setExecutor(...) inside a try/catch (use the withTimeout utility for
any async steps, e.g., when wiring executor calling this.generate), and on any
error log it and skip cron setup (do not call setCronManagerRef or rely on
cronManager afterwards); ensure the code leaves NeuroLink usable even if cron
setup fails and that successful creation still sets the ref and executor on the
CronManager instance.
Comment on lines
+1398
to
+1399
| // Set the reference so cron tools can access the manager | ||
| setCronManagerRef(this.cronManager); |
There was a problem hiding this comment.
Clear cron references during shutdown to avoid stale global manager usage.
Line 1399 stores a shared manager reference (setCronManagerRef(this.cronManager)), but shutdown only closes the instance; it does not clear the local pointer or shared ref. Cron tools can retain and call a shut-down manager after lifecycle transitions.
Proposed lifecycle fix in this file
if (this.cronManager) {
try {
await this.cronManager.shutdown();
+ this.cronManager = undefined;
logger.debug("[NeuroLink] CronManager shutdown completed");
} catch (error) {
logger.warn("[NeuroLink] CronManager shutdown failed:", error);
}
}Please also add a companion clearCronManagerRef() (or nullable setter) in src/lib/agent/directTools.ts and call it from both shutdown() and dispose().
Also applies to: 2249-2256
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib/neurolink.ts` around lines 1398 - 1399, The global cron manager
reference set by setCronManagerRef(this.cronManager) must be cleared during
lifecycle teardown to avoid use-after-shutdown; add a clearCronManagerRef (or a
nullable setter) in src/lib/agent/directTools.ts that nulls the shared ref and
then call that function from both shutdown() and dispose() in
src/lib/neurolink.ts (in the same locations where setCronManagerRef is used,
e.g., the block around setCronManagerRef(this.cronManager) and the
shutdown/dispose implementations) so the shared pointer is cleared whenever the
local manager is closed.
Comment on lines
+2278
to
+2283
| async hasActiveCronTasks(): Promise<boolean> { | ||
| if (!this.cronManager) { | ||
| return false; | ||
| } | ||
| const tasks = await this.cronManager.listTasks({ status: "active" }); | ||
| return tasks.length > 0; |
There was a problem hiding this comment.
Make hasActiveCronTasks() resilient for CLI liveness checks.
If listTasks() throws (store/network/transient failures), this method rejects and can crash the keepalive decision path. Return false on failure with a warning.
Proposed fallback behavior
async hasActiveCronTasks(): Promise<boolean> {
if (!this.cronManager) {
return false;
}
- const tasks = await this.cronManager.listTasks({ status: "active" });
- return tasks.length > 0;
+ try {
+ const tasks = await this.cronManager.listTasks({ status: "active" });
+ return tasks.length > 0;
+ } catch (error) {
+ logger.warn("[NeuroLink] Failed to query active cron tasks", {
+ error: error instanceof Error ? error.message : String(error),
+ });
+ return false;
+ }
}As per coding guidelines, "Use withTimeout utility for async operations and implement graceful degradation with provider fallback for error handling."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib/neurolink.ts` around lines 2278 - 2283, Update hasActiveCronTasks to
guard against exceptions and timeouts from cronManager.listTasks: call
this.cronManager.listTasks(...) wrapped with the withTimeout utility and catch
any errors; on timeout or exception log a warning via the existing logger (e.g.,
processLogger or this.logger) that includes the caught error and return false so
liveness checks degrade gracefully; keep the original success path (return
tasks.length > 0) unchanged and reference hasActiveCronTasks and
this.cronManager.listTasks to locate the change.
Comment on lines
+148
to
+155
| const cronConfig = options?.defaultProvider | ||
| ? { | ||
| cron: { | ||
| defaultProvider: options.defaultProvider, | ||
| defaultModel: options.defaultModel, | ||
| }, | ||
| } | ||
| : undefined; |
There was a problem hiding this comment.
defaultModel is ignored unless defaultProvider is also set.
Current gating only checks options.defaultProvider, so a valid defaultModel-only call silently does nothing.
🔧 Proposed fix
- const cronConfig = options?.defaultProvider
+ const cronConfig = options?.defaultProvider || options?.defaultModel
? {
cron: {
- defaultProvider: options.defaultProvider,
- defaultModel: options.defaultModel,
+ ...(options?.defaultProvider
+ ? { defaultProvider: options.defaultProvider }
+ : {}),
+ ...(options?.defaultModel
+ ? { defaultModel: options.defaultModel }
+ : {}),
},
}
: undefined;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| const cronConfig = options?.defaultProvider | |
| ? { | |
| cron: { | |
| defaultProvider: options.defaultProvider, | |
| defaultModel: options.defaultModel, | |
| }, | |
| } | |
| : undefined; | |
| const cronConfig = options?.defaultProvider || options?.defaultModel | |
| ? { | |
| cron: { | |
| ...(options?.defaultProvider | |
| ? { defaultProvider: options.defaultProvider } | |
| : {}), | |
| ...(options?.defaultModel | |
| ? { defaultModel: options.defaultModel } | |
| : {}), | |
| }, | |
| } | |
| : undefined; |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib/session/globalSessionState.ts` around lines 148 - 155, The cronConfig
construction currently only runs when options.defaultProvider is set so
options.defaultModel is ignored; change the guard to check for either
options.defaultProvider or options.defaultModel (e.g., if
(options?.defaultProvider || options?.defaultModel)) and build the cron object
including defaultProvider and defaultModel only when they exist (use the
existing cronConfig variable and include the fields conditionally) so a call
that supplies defaultModel alone is respected.
Collaborator
|
@swaroopvarma1 review |
Collaborator
|
Verdict: Needs changes — Critical security vulnerabilities must be resolved before merge.
Also worth fixing:
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
from
March 16, 2026 07:49
39d07f2 to
d606f41
Compare
Author
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
Actionable comments posted: 11
♻️ Duplicate comments (15)
src/lib/cron/schedulerBackend.ts (3)
121-128:⚠️ Potential issue | 🟠 MajorCatch rejected scheduler callbacks instead of discarding the Promise.
Each
void callback()here drops the returned Promise. If task execution rejects, the failure escapes the timer boundary as an unhandled rejection instead of being recorded and contained.In Node.js, if an async function is invoked from `setTimeout`, `setInterval`, or another callback and the returned Promise is ignored with `void`, does a rejection surface as an unhandled promise rejection?Based on learnings: "When dealing with backward compatibility concerns for synchronous APIs that internally use async operations, handle Promises with .then()/.catch() to prevent unhandled rejections while preserving the original method signature."
Also applies to: 147-149, 167-169
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/schedulerBackend.ts` around lines 121 - 128, The code currently calls the async callback using `void callback()` inside the timeout branch (and similar places around lines referenced) which drops the returned Promise and can produce unhandled rejections; change these invocations to explicitly handle the Promise (e.g., call `callback().catch(...)` or `Promise.resolve(callback()).catch(...)`) and in the catch record or handle the error via the scheduler's error path (use the same logging/metrics path you use elsewhere in SchedulerBackend, and ensure you still delete `this.timers.delete(task.id)` as needed before/after handling); update all occurrences (the shown `setTimeout` callback plus the other spots mentioned) to consistently catch and record rejections rather than discarding the Promise.
112-149:⚠️ Potential issue | 🔴 CriticalValidate timer delays before handing them to Node.
Line 119 and Line 143 only reject
<= 0, soNaNstill reaches the timer APIs, and"d"intervals can exceed Node’s max supported delay. Both cases end up firing almost immediately instead of on schedule.In Node.js, what happens when `setTimeout` or `setInterval` receive `NaN` or a delay larger than `2147483647`, and what is the maximum supported timer delay?Also applies to: 180-213
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/schedulerBackend.ts` around lines 112 - 149, The scheduleAt and scheduleEvery functions currently only check <= 0, allowing NaN or values > Node's max timer (2147483647) to reach setTimeout/setInterval; update scheduleAt (targetTime/delayMs) and scheduleEvery (intervalMs) to validate that the computed delay/interval is a finite safe integer: if Number.isNaN(delayMs) or !Number.isFinite(delayMs) or delayMs <= 0, reject/throw with a clear error referencing task.id and task.schedule.value; if delayMs/intervalMs > 2147483647, clamp to 2147483647 (or throw consistently with your error handling) before passing to setTimeout/setInterval, and ensure parseInterval results are validated the same way so NaN never reaches timers (refer to scheduleAt, scheduleEvery, timers map and parseInterval).
147-149:⚠️ Potential issue | 🟠 Major
everyschedules can overlap the previous run.
setIntervalkeeps firing on wall-clock time; it does not wait for the async callback. Slow executions will run concurrently and duplicate work, unlike the cron path which already usesprotect: true.Does Node.js `setInterval()` wait for an async callback's Promise to resolve before scheduling the next invocation?🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/schedulerBackend.ts` around lines 147 - 149, The current use of setInterval (creating timer) lets the async callback overlap previous runs; change scheduling so the next invocation waits for callback() to complete: replace the setInterval-based timer with a loop that awaits callback() (or uses a running flag) and then uses setTimeout to schedule the next run after intervalMs. Refer to the existing symbols timer, setInterval, callback, and intervalMs (or the surrounding every schedule logic) and ensure the implementation mimics the cron protect: true behavior by preventing concurrent executions before scheduling the next invocation.src/lib/cron/types.ts (1)
14-238: 🛠️ Refactor suggestion | 🟠 MajorMove this exported cron type surface under
src/lib/types/.These are new shared/public types, but
src/lib/types/configTypes.tsnow has to importCronManagerConfigback from../cron/types.js. That reverses the type-layer dependency and reintroduces the circular-risk this repo’s type layout is meant to prevent.Based on learnings: "all new type definitions must be placed in src/lib/types/" and "Type definitions must be organized by domain to avoid circular dependencies."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/types.ts` around lines 14 - 238, The cron-related exported types (e.g., CronManagerConfig, ScheduledTask, Schedule, TaskRunResult, TaskStore, SchedulerBackend, TaskExecutor, SessionMode, TaskStatus) must be moved into the shared types module under lib/types so they live with other public types and avoid cross-layer imports; create or export them from that shared types module and remove the exports from cron/types.ts, then update any consumers (notably the module currently importing CronManagerConfig from cron/types) to import those symbols from the shared types module instead so no code depends back into the cron layer.src/lib/cron/taskStore.ts (2)
94-113:⚠️ Potential issue | 🟠 MajorPrevent partially initialized Redis clients from escaping
getClient().
this.redisClientis set beforeconnect()completes, so concurrent callers can receive a client that is not ready yet. Memoize an in-flight promise and only publishthis.redisClientafter successful connect.Possible fix
export class RedisTaskStore implements TaskStore { private keyPrefix: string; private redisClient: RedisLikeClient | null = null; + private redisClientPromise: Promise<RedisLikeClient> | null = null; private config: RedisTaskStoreConfig; @@ private async getClient(): Promise<RedisLikeClient> { if (this.redisClient) { return this.redisClient; } + if (this.redisClientPromise) { + return this.redisClientPromise; + } - const { createClient } = await import("redis"); - const url = - this.config.url || - `redis://${this.config.host || "localhost"}:${this.config.port || 6379}`; - this.redisClient = createClient({ - url, - password: this.config.password, - }) as unknown as RedisLikeClient; - await this.redisClient.connect(); - return this.redisClient; + this.redisClientPromise = (async () => { + const { createClient } = await import("redis"); + const url = + this.config.url || + `redis://${this.config.host || "localhost"}:${this.config.port || 6379}`; + const client = createClient({ + url, + password: this.config.password, + }) as unknown as RedisLikeClient; + await client.connect(); + this.redisClient = client; + return client; + })(); + + try { + return await this.redisClientPromise; + } finally { + this.redisClientPromise = null; + } }#!/bin/bash # Verify getClient does not memoize an in-flight connection promise rg -n -C3 'private async getClient|redisClientPromise|createClient|connect\(' src/lib/cron/taskStore.ts🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/taskStore.ts` around lines 94 - 113, getClient currently assigns this.redisClient before connect() completes causing callers to receive an unconnected client; change getClient to memoize an in-flight promise (e.g. this.redisClientPromise) and have concurrent callers await that promise instead of a partially-initialized this.redisClient, only set this.redisClient after createClient(...); await redisClient.connect() succeeds, resolve the promise with the connected client, and ensure the promise is cleared or rejected on connection failure so subsequent calls can retry; update references to getClient, createClient, connect, this.redisClient and add/handle this.redisClientPromise accordingly.
168-187:⚠️ Potential issue | 🟠 MajorMake Redis task updates atomic to avoid lost writes.
updateStatus()andaddRunResult()use read-modify-write (get→ mutate →save). Concurrent updates can overwrite each other (e.g., cancel vs run completion). Use an atomic update path (WATCH/MULTI/EXEC or Lua script) for these transitions.#!/bin/bash # Verify non-atomic read-modify-write patterns in task updates rg -n -C4 'async updateStatus|async addRunResult|await this.get\(|await this.save\(' src/lib/cron/taskStore.ts🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/taskStore.ts` around lines 168 - 187, The updateStatus and addRunResult currently perform non-atomic read-modify-write by calling get(...) then mutating and save(...), which can cause lost updates under concurrency; replace those flows with an atomic Redis update (either a WATCH/MULTI/EXEC transaction or a single Lua script) that reads the task key, applies the mutation (status change, runs.unshift + runCount increment + trimming to maxRunHistory, updatedAt update), and writes back only if the key hasn't changed (or perform the mutation entirely in Lua). Target the functions updateStatus and addRunResult and the usages of get and save to implement the transaction/Lua path so concurrent cancel/complete updates cannot overwrite each other.src/lib/cron/cronManager.ts (3)
57-62:⚠️ Potential issue | 🟠 MajorSilent fallback from Redis to in-memory store loses user data.
When
store: "redis"is configured butredisConfigis missing, the code silently falls back toInMemoryTaskStore. This violates user expectations—they believe tasks are persisted but they vanish on process restart.Either throw a configuration error or log a prominent warning:
🔧 Proposed fix
// Initialize store if (this.config.store === "redis" && this.config.redisConfig) { this.store = new RedisTaskStore(this.config.redisConfig); + } else if (this.config.store === "redis") { + throw new Error( + "Cron store is set to 'redis' but redisConfig is missing. " + + "Provide redisConfig or set store to 'memory'." + ); } else { this.store = new InMemoryTaskStore(); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/cronManager.ts` around lines 57 - 62, The constructor/initialization currently silently uses InMemoryTaskStore when this.config.store === "redis" but this.config.redisConfig is missing, which causes data loss; update the initialization logic in cronManager (where RedisTaskStore and InMemoryTaskStore are chosen) to detect this invalid configuration and fail-fast: if this.config.store === "redis" and !this.config.redisConfig, throw a clear ConfigurationError (or similar) with an explanatory message (or at minimum log an explicit error and throw) instead of falling back to InMemoryTaskStore so users are notified and persistence is not silently lost.
286-293:⚠️ Potential issue | 🟠 MajorWrap executor call with
withTimeoutto prevent indefinite blocking.Line 287 awaits
this.executor(task, sessionId)with no timeout. A hung model call can hold a concurrency slot forever, blocking subsequent task executions and potentially causing resource exhaustion.As per coding guidelines: "Use withTimeout utility for async operations and implement graceful degradation with provider fallback for error handling."
🔧 Proposed fix
+import { withTimeout } from "../utils/timeout.js"; + +// Default execution timeout (5 minutes) +const DEFAULT_EXECUTION_TIMEOUT_MS = 5 * 60 * 1000; try { - const result = await this.executor(task, sessionId); + const executionTimeout = this.config.executionTimeoutMs ?? DEFAULT_EXECUTION_TIMEOUT_MS; + const result = await withTimeout( + this.executor(task, sessionId), + executionTimeout, + `Task execution timed out after ${executionTimeout}ms` + ); run.status = "completed";🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/cronManager.ts` around lines 286 - 293, The call to this.executor(task, sessionId) can hang indefinitely; wrap it with the withTimeout utility (using the configured timeout constant or a sensible default) so the executor promise is rejected if it exceeds the timeout, then handle the timeout/rejection in the existing catch to mark run.status = "failed", set run.completedAt/run.durationMs, record the error message on run (and clear or set run.responseText/tokenUsage appropriately), and trigger any provider-fallback logic per the guidelines; update references around this.executor, withTimeout, and the run object in cronManager.ts to implement this change.
138-162:⚠️ Potential issue | 🟠 MajorValidate
creatorSessionIdwhensessionModeis"same-session".When
sessionModeis"same-session"butcreatorSessionIdis not provided, the code silently falls back tocron:${taskId}at line 268. This defeats the purpose of "same-session" mode—the task runs in an isolated session anyway.🔧 Proposed fix
async createTask(options: CreateTaskOptions): Promise<ScheduledTask> { if (!this.config.enabled) { throw new Error("Cron system is disabled"); } if (this.isShutdown) { throw new Error("CronManager is shut down"); } + if ( + (options.sessionMode ?? "isolated") === "same-session" && + !options.creatorSessionId + ) { + throw new Error( + '"same-session" mode requires creatorSessionId to be provided' + ); + } const task: ScheduledTask = {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/cronManager.ts` around lines 138 - 162, In createTask, enforce that when the ScheduledTask.sessionMode is "same-session" the options.creatorSessionId must be provided: check options.sessionMode (or task.sessionMode) === "same-session" and if so throw a clear Error if options.creatorSessionId is missing instead of allowing a fallback; ensure the ScheduledTask.creatorSessionId is populated from options.creatorSessionId only and remove/avoid the later silent assignment to `cron:${taskId}` so "same-session" truly runs in the provided session.src/lib/neurolink.ts (6)
1354-1362:⚠️ Potential issue | 🟠 MajorValidate env-derived cron config before assigning.
storeis force-cast andmaxConcurrentRunscan becomeNaN, which can leak invalid runtime config intoCronManager.Proposed hardening
- const envStore = process.env.NEUROLINK_CRON_STORE; - const envMaxConcurrent = process.env.NEUROLINK_CRON_MAX_CONCURRENT; + const envStoreRaw = process.env.NEUROLINK_CRON_STORE; + const envMaxConcurrentRaw = process.env.NEUROLINK_CRON_MAX_CONCURRENT; + + const envStore = + envStoreRaw === "memory" || envStoreRaw === "redis" + ? envStoreRaw + : undefined; + + const envMaxConcurrent = (() => { + if (!envMaxConcurrentRaw) return undefined; + const parsed = Number.parseInt(envMaxConcurrentRaw, 10); + return Number.isInteger(parsed) && parsed > 0 ? parsed : undefined; + })(); @@ - store: (envStore as "memory" | "redis") || cronConfig?.store || "memory", - maxConcurrentRuns: envMaxConcurrent - ? parseInt(envMaxConcurrent, 10) - : cronConfig?.maxConcurrentRuns, + store: envStore ?? cronConfig?.store ?? "memory", + maxConcurrentRuns: envMaxConcurrent ?? cronConfig?.maxConcurrentRuns,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/neurolink.ts` around lines 1354 - 1362, The env-derived cron config is not validated: envStore is force-cast and envMaxConcurrent can produce NaN, allowing invalid values into CronManagerConfig (config). Fix by validating and normalizing before assignment: for store, only accept "memory" or "redis" (fall back to cronConfig?.store or "memory" if invalid); for maxConcurrentRuns, parse envMaxConcurrent to an integer and only use it if Number.isFinite and > 0, otherwise fall back to cronConfig?.maxConcurrentRuns or a safe default. Update the construction of config (CronManagerConfig) to use these validated variables instead of directly using envStore/envMaxConcurrent.
1445-1449:⚠️ Potential issue | 🔴 CriticalBound cron executor generation with timeout.
this.generate(...)can run indefinitely; one hung run can stall scheduler throughput.Proposed fix
- let result; + let result: GenerateResult; try { - result = await this.generate( - generateOptions as import("./types/generateTypes.js").GenerateOptions, - ); + result = await withTimeout( + this.generate( + generateOptions as import("./types/generateTypes.js").GenerateOptions, + ), + 120000, + new Error(`Cron task ${task.id} timed out`), + ); logger.debug(`[CronExecutor] Task ${task.id} completed successfully`); } catch (error) {As per coding guidelines, "Use withTimeout utility for async operations and implement graceful degradation with provider fallback for error handling."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/neurolink.ts` around lines 1445 - 1449, The call to this.generate(...) can hang and must be wrapped with the withTimeout utility: replace the direct await this.generate(generateOptions as GenerateOptions) with await withTimeout(this.generate(generateOptions as import("./types/generateTypes.js").GenerateOptions), timeoutMs) and catch timeout errors separately; on timeout or other generation errors implement graceful degradation by attempting the next provider in your provider list (use the module/class that manages providers, e.g., this.providers or the provider selection method) and only surface a final failure after all fallbacks are exhausted, ensuring you log context and return/throw a controlled error state rather than letting the scheduler stall.
1370-1475:⚠️ Potential issue | 🟠 MajorCron init failure can break
NeuroLinkconstruction.
new CronManager(config)and wiring are unguarded; any backend/init error can throw from constructor and disable the entire SDK instance instead of degrading gracefully.As per coding guidelines, "Use withTimeout utility for async operations and implement graceful degradation with provider fallback for error handling." and "Maintain backward compatibility with existing SDK APIs when making changes. All SDK modifications must not break existing code using the library."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/neurolink.ts` around lines 1370 - 1475, Wrap CronManager construction and the subsequent wiring (new CronManager(config), this.cronManager.setExecutor(...), and setCronManagerRef(this.cronManager)) in a try/catch so initialization failures don't throw from NeuroLink's constructor; on error log the failure and assign a safe fallback (e.g., this.cronManager = null or a NoopCronManager implementing the same API) and call setCronManagerRef with that fallback so callers still work; for any async init within the executor or CronManager creation use the existing withTimeout utility when awaiting external/init operations to avoid hung startups; reference CronManager, setExecutor, setCronManagerRef, and withTimeout when locating the changes.
2348-2353:⚠️ Potential issue | 🟡 MinorMake
hasActiveCronTasks()fail-safe for liveness checks.
listTasks()errors currently reject this method. Wrap with timeout + catch and returnfalseon failure.Proposed fix
async hasActiveCronTasks(): Promise<boolean> { if (!this.cronManager) { return false; } - const tasks = await this.cronManager.listTasks({ status: "active" }); - return tasks.length > 0; + try { + const tasks = await withTimeout( + this.cronManager.listTasks({ status: "active" }), + 3000, + new Error("Cron active-task check timed out"), + ); + return tasks.length > 0; + } catch (error) { + logger.warn("[NeuroLink] Failed to query active cron tasks", { + error: error instanceof Error ? error.message : String(error), + }); + return false; + } }As per coding guidelines, "Use withTimeout utility for async operations and implement graceful degradation with provider fallback for error handling."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/neurolink.ts` around lines 2348 - 2353, hasActiveCronTasks currently lets listTasks rejections bubble up; wrap the call to this.cronManager.listTasks({ status: "active" }) with the project's withTimeout utility and a try/catch so any timeout or error returns false (preserving the existing early return when !this.cronManager); update the method (hasActiveCronTasks) to await withTimeout(...) and on any thrown error or timeout catch it and return false to make the liveness check fail-safe.
1468-1469:⚠️ Potential issue | 🟠 MajorClear cron manager references during teardown.
You set a shared manager ref, but shutdown path doesn’t clear shared/local pointers after closing. That risks stale manager usage post-shutdown (and dispose path should also handle cron cleanup consistently).
Also applies to: 2319-2326
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/neurolink.ts` around lines 1468 - 1469, The code sets a shared cron manager via setCronManagerRef(this.cronManager) but never clears that shared or local pointer on shutdown/dispose, which can leave stale references; update the teardown paths (the shutdown and dispose code paths that close/stop the cron manager) to explicitly clear both the shared reference and the instance field—call setCronManagerRef(null) after successfully closing/stopping this.cronManager and then set this.cronManager = null (and do the same in the dispose method/path) so no stale manager remains accessible after shutdown.
1403-1439:⚠️ Potential issue | 🔴 CriticalDo not enable all direct tools for scheduled prompts by default.
getToolsForCategory("all")gives cron prompts broad tool access and expands blast radius for prompt injection/abuse. Restrict this to an explicit allowlist per task/config.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/neurolink.ts` around lines 1403 - 1439, Replace the blanket call to getToolsForCategory("all") and the unconditional registration/usage of those tools with an explicit allowlist per task or config: read allowed tool names from task.allowedTools (falling back to config.allowedTools or an empty list), build a filtered tools object by selecting only entries from getToolsForCategory for those names, register only those filtered tools using registerTool (keep the same registration logic around toolDef/execute/inputSchema), and pass the filtered tools into generateOptions instead of the unrestricted tools variable; ensure symbols referenced are getToolsForCategory, tools, registerTool, task.allowedTools, config.allowedTools, and generateOptions.
🧹 Nitpick comments (4)
src/lib/cron/types.ts (1)
161-195: Prefertypealiases for these new exported contracts.
SchedulerBackendandTaskStoreare new shared types, and the project standard is to usetypefor new definitions unless there is a concrete interface-only need.Based on learnings: "new type definitions should use the
typekeyword instead ofinterface, unless there is a valid and justified exception."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/types.ts` around lines 161 - 195, The exported contracts SchedulerBackend and TaskStore are declared as interfaces but the project convention prefers type aliases for new shared types; change both declarations from "interface SchedulerBackend" and "interface TaskStore" to exported type aliases (e.g., "export type SchedulerBackend = { ... }" and "export type TaskStore = { ... }") keeping the same method signatures and return types (schedule, cancel, cancelAll, getNextRunTime, shutdown for SchedulerBackend; save, get, list, delete, updateStatus, addRunResult, shutdown for TaskStore) so callers and implementations remain unchanged.src/lib/core/modules/ToolsManager.ts (1)
149-149: Snapshot direct tools once per call path.
this.directToolsGetter()is invoked repeatedly in the same method flow. Cache one local snapshot and reuse it to avoid repeated recomputation and inconsistent counts when the getter is dynamic.Also applies to: 221-234
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/core/modules/ToolsManager.ts` at line 149, The code calls this.directToolsGetter() multiple times causing repeated recomputation and inconsistent counts; capture a single snapshot by calling this.directToolsGetter() once into a local variable (e.g., const directTools = this.directToolsGetter()) at the start of the method and then use directTools for getKeyCount and any subsequent accesses (replace other direct calls between the directToolsCount line and the block around lines 221-234 with the local variable), ensuring all uses reference the same snapshot.src/lib/agent/directTools.ts (1)
919-925: Tighten cron tools return type to keep strict TS guarantees.
getCronTools()currently returnsRecord<string, any>, which weakens type safety in the tool pipeline. ReturnRecord<string, Tool>and drop the explicit-any suppression.Suggested change
+import type { Tool } from "ai"; ... -// eslint-disable-next-line `@typescript-eslint/no-explicit-any` -export function getCronTools(): Record<string, any> { +export function getCronTools(): Record<string, Tool> { if (shouldDisableCronTools()) { return {}; } return createCronTools(() => cronManagerRef); }As per coding guidelines, "Maintain strict TypeScript across all modules with no circular dependencies between type definition files."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/agent/directTools.ts` around lines 919 - 925, getCronTools currently weakens type safety by returning Record<string, any>; change its signature to return Record<string, Tool> and remove the "// eslint-disable-next-line `@typescript-eslint/no-explicit-any`" suppression. Ensure you import the Tool type (or reference the existing Tool type alias) and adjust the return sites: when shouldDisableCronTools() is true return an empty object typed as Record<string, Tool> and otherwise return the result of createCronTools(() => cronManagerRef) which should already be compatible; verify or adapt createCronTools' typing if necessary while avoiding circular type-imports.src/lib/cron/cronManager.ts (1)
96-109: Synced tasks from BullMQ have empty prompts and cannot re-execute.Tasks reconstructed from BullMQ (line 101:
prompt: "") will appear in listings but fail silently or produce empty outputs if executed. Consider either:
- Storing prompts in BullMQ job data so they can be recovered
- Marking synced tasks with a flag indicating they're orphaned/view-only
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/lib/cron/cronManager.ts` around lines 96 - 109, The reconstructed ScheduledTask created in cronManager.ts is losing its prompt (prompt: "") so synced tasks cannot re-execute; fix by reading the original prompt from the BullMQ job payload (use bullMQTask.data.prompt or the appropriate property on bullMQTask.data) when building the ScheduledTask in the block that constructs task (refer to ScheduledTask and bullMQTask.id/name/schedule), and if the job data has no prompt, mark the task as orphaned/view-only by adding a flag (e.g., isOrphan or source = "bullmq-orphan") or set status to a dedicated value so the UI/runner knows it cannot execute; ensure ScheduledTask type is updated to include the flag if needed and that the loader uses bullMQTask.data.prompt as the primary source before falling back to orphan handling.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@scripts/bullmq-dashboard.ts`:
- Around line 17-18: The PORT constant (PORT) is currently a string which causes
app.listen() to treat it as a socket path; parse the environment value into a
number (e.g., via Number(...) or parseInt(..., 10)) and validate it before
passing to app.listen() so the server binds to a TCP port; update the code that
uses PORT (where app.listen(...) is called) to pass the parsed numeric value
(and handle NaN/fallback to 3000 if needed).
In `@src/lib/cron/bullmqScheduler.ts`:
- Around line 57-62: The getInstance function currently always returns the first
created BullMQScheduler stored on globalThis.__bullMQScheduler, causing
subsequent calls with different redisUrl to reuse the wrong Redis connection;
change the singleton strategy so schedulers are keyed by their Redis
configuration (e.g., maintain a global map keyed by redisUrl or a normalized
connection string) and return or create a BullMQScheduler for the requested
redisUrl (or alternatively remove the cross-process singleton so new
BullMQScheduler(redisUrl) is always created); update references to getInstance
and the global storage (__bullMQScheduler) to use this keyed lookup so each
distinct redisUrl gets its own BullMQScheduler instance.
- Line 47: The in-memory Map<string, () => Promise<void>> callbacks is lost on
restart so ready jobs get dropped; persist callback identifiers and handlers to
a durable store (e.g., Redis) and ensure they are rehydrated before the worker
starts processing. Concretely: when registering a callback (the existing
callbacks Map usage), store a stable callbackId and the handler metadata in
Redis (or attach callbackId to job.data), update the register function to write
that persistent mapping, implement a rehydrateCallbacks function that loads all
callbackIds/handlers from Redis into callbacks on startup, and ensure the
Worker/processing loop (the code that currently logs "No callback found") is
paused or not started until rehydrateCallbacks completes (or have the job
handler look up the persisted callback by callbackId at process time). This
preserves BullMQ persistence across restarts and prevents jobs from being
dropped.
- Around line 96-112: The initializeBullMQ startup currently awaits
Queue.waitUntilReady() (and similarly Worker.waitUntilReady() elsewhere) with no
timeout, which can hang initialization; update initializeBullMQ() to wrap
this.queue.waitUntilReady() in the existing withTimeout() utility (use a 10–15s
timeout) and do the same for Worker.waitUntilReady() calls so failures fall
through to the existing fallback logic that constructs a NodeTimeoutScheduler;
ensure the withTimeout call rejects on timeout so the catch path triggers the
provider fallback.
In `@src/lib/cron/cronManager.ts`:
- Around line 264-269: The fallback that sets sessionId to `cron:${taskId}` when
`task.sessionMode === "same-session"` causes same-session to behave like
isolated; change the logic so `sessionId` is only generated for `"isolated"` and
for `"same-session"` it must use `task.creatorSessionId` (do not fallback to a
generated id), and add/ensure validation in `createTask` to throw or reject when
`sessionMode === "same-session"` but `creatorSessionId` is undefined; update
references to `runNumber`, `sessionId`, `task.sessionMode`,
`task.creatorSessionId`, and `createTask` accordingly.
- Around line 138-184: createTask currently allows unlimited task creation which
can exhaust storage; add a new config field (e.g., this.config.maxTasks) and
enforce it at the top of createTask by checking the current count (via
this.store.countTasks or this.store.listKeys) and throwing a clear error when
the limit is reached, and add a TTL/auto-prune behavior for completed tasks by
setting an expiry or scheduling cleanup in the store layer (e.g., when updating
task.status to "completed" in methods like executeTask or this.store.save) using
this.config.completedTaskTTL so old completed tasks are removed automatically;
update related places that create or persist tasks (createTask, executeTask, and
store.save) to honor these config values and emit a log when a task is rejected
due to maxTasks.
In `@src/lib/cron/taskStore.ts`:
- Around line 25-32: The in-memory task store (methods save, get and list in
taskStore.ts) currently uses shallow clones so nested/mutable fields like
ScheduledTask.runs remain shared; update save to store a deep-cloned copy of the
incoming ScheduledTask and update get and list to return deep-cloned copies so
callers cannot mutate internal state; use a reliable deep-clone utility (e.g.,
structuredClone or a safe deep-copy helper) to clone the whole ScheduledTask
object both when persisting in save and when returning values from get and list.
- Around line 123-128: Extend the RedisLikeClient interface to accept TTL
options (e.g., set(key: string, value: string, options?: { EX?: number })) and
update save(task: ScheduledTask) in taskStore to pass an appropriate expiry when
persisting tasks: determine TTL from task.status (e.g., use retentionCompleted
for completed/cancelled tasks, retentionDefault for active tasks) and call
client.set(this.taskKey(task.id), JSON.stringify(task), { EX: ttlInSeconds });
ensure existing uses of sAdd(this.indexKey(), task.id) remain unchanged and make
retention values configurable (e.g., class ctor or config object) so they can be
tuned without code changes.
In `@src/lib/neurolink.ts`:
- Around line 1379-1380: The code that builds previousRuns uses
task.runs.slice(0, 3) which returns the first three runs, not the latest three;
update the extraction of the last three runs (e.g., use task.runs.slice(-3) or
otherwise reverse/sort by timestamp before slicing) where previousRuns is
assigned so the variable truly contains the most recent runs for subsequent
logic in this function.
- Line 1445: The local variable "result" is declared without a type which breaks
strict TypeScript; update the declaration to include the explicit return type
from this.generate() by changing the untyped declaration to use GenerateResult
(i.e., declare result as type GenerateResult). Locate the variable named result
in the method where this.generate() is called (in src/lib/neurolink.ts) and
change its declaration to include the GenerateResult type so the compiler knows
the expected shape.
In `@src/lib/session/globalSessionState.ts`:
- Around line 149-159: The code hardcodes "litellm" and "kimi-latest" into the
NeuroLink cron defaults, which overrides the SDK's normal provider/model
resolution; update the NeuroLink instantiation in globalSessionState.ts to avoid
forcing defaults when callers only provide defaultModel or nothing—pass through
options?.defaultProvider and options?.defaultModel as-is (or omit those keys
entirely when undefined) so NeuroLink's/SDK's resolver retains responsibility
for choosing the provider/model; modify the cron object construction around the
cronProvider/cronModel usage to only set defaultProvider/defaultModel when the
corresponding option is explicitly provided.
---
Duplicate comments:
In `@src/lib/cron/cronManager.ts`:
- Around line 57-62: The constructor/initialization currently silently uses
InMemoryTaskStore when this.config.store === "redis" but this.config.redisConfig
is missing, which causes data loss; update the initialization logic in
cronManager (where RedisTaskStore and InMemoryTaskStore are chosen) to detect
this invalid configuration and fail-fast: if this.config.store === "redis" and
!this.config.redisConfig, throw a clear ConfigurationError (or similar) with an
explanatory message (or at minimum log an explicit error and throw) instead of
falling back to InMemoryTaskStore so users are notified and persistence is not
silently lost.
- Around line 286-293: The call to this.executor(task, sessionId) can hang
indefinitely; wrap it with the withTimeout utility (using the configured timeout
constant or a sensible default) so the executor promise is rejected if it
exceeds the timeout, then handle the timeout/rejection in the existing catch to
mark run.status = "failed", set run.completedAt/run.durationMs, record the error
message on run (and clear or set run.responseText/tokenUsage appropriately), and
trigger any provider-fallback logic per the guidelines; update references around
this.executor, withTimeout, and the run object in cronManager.ts to implement
this change.
- Around line 138-162: In createTask, enforce that when the
ScheduledTask.sessionMode is "same-session" the options.creatorSessionId must be
provided: check options.sessionMode (or task.sessionMode) === "same-session" and
if so throw a clear Error if options.creatorSessionId is missing instead of
allowing a fallback; ensure the ScheduledTask.creatorSessionId is populated from
options.creatorSessionId only and remove/avoid the later silent assignment to
`cron:${taskId}` so "same-session" truly runs in the provided session.
In `@src/lib/cron/schedulerBackend.ts`:
- Around line 121-128: The code currently calls the async callback using `void
callback()` inside the timeout branch (and similar places around lines
referenced) which drops the returned Promise and can produce unhandled
rejections; change these invocations to explicitly handle the Promise (e.g.,
call `callback().catch(...)` or `Promise.resolve(callback()).catch(...)`) and in
the catch record or handle the error via the scheduler's error path (use the
same logging/metrics path you use elsewhere in SchedulerBackend, and ensure you
still delete `this.timers.delete(task.id)` as needed before/after handling);
update all occurrences (the shown `setTimeout` callback plus the other spots
mentioned) to consistently catch and record rejections rather than discarding
the Promise.
- Around line 112-149: The scheduleAt and scheduleEvery functions currently only
check <= 0, allowing NaN or values > Node's max timer (2147483647) to reach
setTimeout/setInterval; update scheduleAt (targetTime/delayMs) and scheduleEvery
(intervalMs) to validate that the computed delay/interval is a finite safe
integer: if Number.isNaN(delayMs) or !Number.isFinite(delayMs) or delayMs <= 0,
reject/throw with a clear error referencing task.id and task.schedule.value; if
delayMs/intervalMs > 2147483647, clamp to 2147483647 (or throw consistently with
your error handling) before passing to setTimeout/setInterval, and ensure
parseInterval results are validated the same way so NaN never reaches timers
(refer to scheduleAt, scheduleEvery, timers map and parseInterval).
- Around line 147-149: The current use of setInterval (creating timer) lets the
async callback overlap previous runs; change scheduling so the next invocation
waits for callback() to complete: replace the setInterval-based timer with a
loop that awaits callback() (or uses a running flag) and then uses setTimeout to
schedule the next run after intervalMs. Refer to the existing symbols timer,
setInterval, callback, and intervalMs (or the surrounding every schedule logic)
and ensure the implementation mimics the cron protect: true behavior by
preventing concurrent executions before scheduling the next invocation.
In `@src/lib/cron/taskStore.ts`:
- Around line 94-113: getClient currently assigns this.redisClient before
connect() completes causing callers to receive an unconnected client; change
getClient to memoize an in-flight promise (e.g. this.redisClientPromise) and
have concurrent callers await that promise instead of a partially-initialized
this.redisClient, only set this.redisClient after createClient(...); await
redisClient.connect() succeeds, resolve the promise with the connected client,
and ensure the promise is cleared or rejected on connection failure so
subsequent calls can retry; update references to getClient, createClient,
connect, this.redisClient and add/handle this.redisClientPromise accordingly.
- Around line 168-187: The updateStatus and addRunResult currently perform
non-atomic read-modify-write by calling get(...) then mutating and save(...),
which can cause lost updates under concurrency; replace those flows with an
atomic Redis update (either a WATCH/MULTI/EXEC transaction or a single Lua
script) that reads the task key, applies the mutation (status change,
runs.unshift + runCount increment + trimming to maxRunHistory, updatedAt
update), and writes back only if the key hasn't changed (or perform the mutation
entirely in Lua). Target the functions updateStatus and addRunResult and the
usages of get and save to implement the transaction/Lua path so concurrent
cancel/complete updates cannot overwrite each other.
In `@src/lib/cron/types.ts`:
- Around line 14-238: The cron-related exported types (e.g., CronManagerConfig,
ScheduledTask, Schedule, TaskRunResult, TaskStore, SchedulerBackend,
TaskExecutor, SessionMode, TaskStatus) must be moved into the shared types
module under lib/types so they live with other public types and avoid
cross-layer imports; create or export them from that shared types module and
remove the exports from cron/types.ts, then update any consumers (notably the
module currently importing CronManagerConfig from cron/types) to import those
symbols from the shared types module instead so no code depends back into the
cron layer.
In `@src/lib/neurolink.ts`:
- Around line 1354-1362: The env-derived cron config is not validated: envStore
is force-cast and envMaxConcurrent can produce NaN, allowing invalid values into
CronManagerConfig (config). Fix by validating and normalizing before assignment:
for store, only accept "memory" or "redis" (fall back to cronConfig?.store or
"memory" if invalid); for maxConcurrentRuns, parse envMaxConcurrent to an
integer and only use it if Number.isFinite and > 0, otherwise fall back to
cronConfig?.maxConcurrentRuns or a safe default. Update the construction of
config (CronManagerConfig) to use these validated variables instead of directly
using envStore/envMaxConcurrent.
- Around line 1445-1449: The call to this.generate(...) can hang and must be
wrapped with the withTimeout utility: replace the direct await
this.generate(generateOptions as GenerateOptions) with await
withTimeout(this.generate(generateOptions as
import("./types/generateTypes.js").GenerateOptions), timeoutMs) and catch
timeout errors separately; on timeout or other generation errors implement
graceful degradation by attempting the next provider in your provider list (use
the module/class that manages providers, e.g., this.providers or the provider
selection method) and only surface a final failure after all fallbacks are
exhausted, ensuring you log context and return/throw a controlled error state
rather than letting the scheduler stall.
- Around line 1370-1475: Wrap CronManager construction and the subsequent wiring
(new CronManager(config), this.cronManager.setExecutor(...), and
setCronManagerRef(this.cronManager)) in a try/catch so initialization failures
don't throw from NeuroLink's constructor; on error log the failure and assign a
safe fallback (e.g., this.cronManager = null or a NoopCronManager implementing
the same API) and call setCronManagerRef with that fallback so callers still
work; for any async init within the executor or CronManager creation use the
existing withTimeout utility when awaiting external/init operations to avoid
hung startups; reference CronManager, setExecutor, setCronManagerRef, and
withTimeout when locating the changes.
- Around line 2348-2353: hasActiveCronTasks currently lets listTasks rejections
bubble up; wrap the call to this.cronManager.listTasks({ status: "active" })
with the project's withTimeout utility and a try/catch so any timeout or error
returns false (preserving the existing early return when !this.cronManager);
update the method (hasActiveCronTasks) to await withTimeout(...) and on any
thrown error or timeout catch it and return false to make the liveness check
fail-safe.
- Around line 1468-1469: The code sets a shared cron manager via
setCronManagerRef(this.cronManager) but never clears that shared or local
pointer on shutdown/dispose, which can leave stale references; update the
teardown paths (the shutdown and dispose code paths that close/stop the cron
manager) to explicitly clear both the shared reference and the instance
field—call setCronManagerRef(null) after successfully closing/stopping
this.cronManager and then set this.cronManager = null (and do the same in the
dispose method/path) so no stale manager remains accessible after shutdown.
- Around line 1403-1439: Replace the blanket call to getToolsForCategory("all")
and the unconditional registration/usage of those tools with an explicit
allowlist per task or config: read allowed tool names from task.allowedTools
(falling back to config.allowedTools or an empty list), build a filtered tools
object by selecting only entries from getToolsForCategory for those names,
register only those filtered tools using registerTool (keep the same
registration logic around toolDef/execute/inputSchema), and pass the filtered
tools into generateOptions instead of the unrestricted tools variable; ensure
symbols referenced are getToolsForCategory, tools, registerTool,
task.allowedTools, config.allowedTools, and generateOptions.
---
Nitpick comments:
In `@src/lib/agent/directTools.ts`:
- Around line 919-925: getCronTools currently weakens type safety by returning
Record<string, any>; change its signature to return Record<string, Tool> and
remove the "// eslint-disable-next-line `@typescript-eslint/no-explicit-any`"
suppression. Ensure you import the Tool type (or reference the existing Tool
type alias) and adjust the return sites: when shouldDisableCronTools() is true
return an empty object typed as Record<string, Tool> and otherwise return the
result of createCronTools(() => cronManagerRef) which should already be
compatible; verify or adapt createCronTools' typing if necessary while avoiding
circular type-imports.
In `@src/lib/core/modules/ToolsManager.ts`:
- Line 149: The code calls this.directToolsGetter() multiple times causing
repeated recomputation and inconsistent counts; capture a single snapshot by
calling this.directToolsGetter() once into a local variable (e.g., const
directTools = this.directToolsGetter()) at the start of the method and then use
directTools for getKeyCount and any subsequent accesses (replace other direct
calls between the directToolsCount line and the block around lines 221-234 with
the local variable), ensuring all uses reference the same snapshot.
In `@src/lib/cron/cronManager.ts`:
- Around line 96-109: The reconstructed ScheduledTask created in cronManager.ts
is losing its prompt (prompt: "") so synced tasks cannot re-execute; fix by
reading the original prompt from the BullMQ job payload (use
bullMQTask.data.prompt or the appropriate property on bullMQTask.data) when
building the ScheduledTask in the block that constructs task (refer to
ScheduledTask and bullMQTask.id/name/schedule), and if the job data has no
prompt, mark the task as orphaned/view-only by adding a flag (e.g., isOrphan or
source = "bullmq-orphan") or set status to a dedicated value so the UI/runner
knows it cannot execute; ensure ScheduledTask type is updated to include the
flag if needed and that the loader uses bullMQTask.data.prompt as the primary
source before falling back to orphan handling.
In `@src/lib/cron/types.ts`:
- Around line 161-195: The exported contracts SchedulerBackend and TaskStore are
declared as interfaces but the project convention prefers type aliases for new
shared types; change both declarations from "interface SchedulerBackend" and
"interface TaskStore" to exported type aliases (e.g., "export type
SchedulerBackend = { ... }" and "export type TaskStore = { ... }") keeping the
same method signatures and return types (schedule, cancel, cancelAll,
getNextRunTime, shutdown for SchedulerBackend; save, get, list, delete,
updateStatus, addRunResult, shutdown for TaskStore) so callers and
implementations remain unchanged.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: fa9221bf-788c-40c9-a9a2-2be4b0685bc2
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (20)
package.jsonscripts/bullmq-dashboard.tssrc/cli/factories/commandFactory.tssrc/lib/agent/directTools.tssrc/lib/core/baseProvider.tssrc/lib/core/modules/ToolsManager.tssrc/lib/cron/bullmqScheduler.tssrc/lib/cron/cronManager.tssrc/lib/cron/cronTools.tssrc/lib/cron/index.tssrc/lib/cron/schedulerBackend.tssrc/lib/cron/taskStore.tssrc/lib/cron/types.tssrc/lib/neurolink.tssrc/lib/providers/googleAiStudio.tssrc/lib/providers/googleVertex.tssrc/lib/session/globalSessionState.tssrc/lib/types/configTypes.tssrc/lib/types/index.tssrc/lib/utils/toolUtils.ts
🚧 Files skipped from review as they are similar to previous changes (4)
- src/lib/providers/googleAiStudio.ts
- src/lib/types/index.ts
- src/lib/providers/googleVertex.ts
- src/lib/cron/cronTools.ts
Comment on lines
+96
to
+112
| private async initializeBullMQ(): Promise<void> { | ||
| try { | ||
| this.queue = new Queue<CronJobData>("neurolink-cron", { | ||
| connection: { url: this.redisUrl }, | ||
| defaultJobOptions: { | ||
| attempts: 3, | ||
| backoff: { | ||
| type: "exponential", | ||
| delay: 1000, | ||
| }, | ||
| removeOnComplete: { count: 100 }, | ||
| removeOnFail: { count: 50 }, | ||
| }, | ||
| }); | ||
|
|
||
| // Test connection | ||
| await this.queue.waitUntilReady(); |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Does BullMQ Queue.waitUntilReady()/Worker.waitUntilReady() have a built-in timeout, or can they wait indefinitely while reconnecting to Redis?
💡 Result:
No—Queue.waitUntilReady() / Worker.waitUntilReady() do not have a built-in timeout.
QueueBase.waitUntilReady()just returnsthis.client(a promise for the underlying Redis connection) and does not apply any timer/limit itself. (raw.githubusercontent.com)Worker.waitUntilReady()simply awaitssuper.waitUntilReady()and then returns the blocking connection’s client—again, no timeout. (raw.githubusercontent.com)- Under the hood, BullMQ’s
RedisConnection.waitUntilReady(client)waits for Redis to emitready(or rejects onend). There is no timeout in this wait logic, so if ioredis keeps reconnecting, this can wait indefinitely. (raw.githubusercontent.com)
Citations:
- 1: https://raw.githubusercontent.com/taskforcesh/bullmq/ddc089799f23c010b00fcb15716dc351f3ad170f/src/classes/queue-base.ts
- 2: https://raw.githubusercontent.com/taskforcesh/bullmq/ddc089799f23c010b00fcb15716dc351f3ad170f/src/classes/worker.ts
- 3: https://raw.githubusercontent.com/taskforcesh/bullmq/ddc089799f23c010b00fcb15716dc351f3ad170f/src/classes/redis-connection.ts
🏁 Script executed:
# First, let's locate and read the bullmqScheduler.ts file
find . -name "bullmqScheduler.ts" -type fRepository: juspay/neurolink
Length of output: 93
🏁 Script executed:
# Read the entire file to understand the context
cat -n ./src/lib/cron/bullmqScheduler.tsRepository: juspay/neurolink
Length of output: 22639
🏁 Script executed:
# Search for withTimeout utility
rg "withTimeout" --type tsRepository: juspay/neurolink
Length of output: 6768
🏁 Script executed:
# Read the withTimeout implementation
cat -n ./src/lib/utils/async/withTimeout.tsRepository: juspay/neurolink
Length of output: 3739
Wrap BullMQ readiness checks with withTimeout() to enable fallback to NodeTimeoutScheduler.
Queue.waitUntilReady() and Worker.waitUntilReady() have no built-in timeout—they wait indefinitely for Redis readiness. A blackholed Redis endpoint will hang initPromise forever, blocking schedule() from ever reaching the fallback mechanism.
Apply withTimeout() to lines 112, 183, and any other critical async calls that could stall initialization. A recommended timeout is 10-15 seconds to allow graceful degradation. This aligns with the coding guideline: "Use withTimeout utility for async operations and implement graceful degradation with provider fallback for error handling."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib/cron/bullmqScheduler.ts` around lines 96 - 112, The initializeBullMQ
startup currently awaits Queue.waitUntilReady() (and similarly
Worker.waitUntilReady() elsewhere) with no timeout, which can hang
initialization; update initializeBullMQ() to wrap this.queue.waitUntilReady() in
the existing withTimeout() utility (use a 10–15s timeout) and do the same for
Worker.waitUntilReady() calls so failures fall through to the existing fallback
logic that constructs a NodeTimeoutScheduler; ensure the withTimeout call
rejects on timeout so the catch path triggers the provider fallback.
| `[CronExecutor] Executing task ${task.id} with ${effectiveProvider}/${effectiveModel}`, | ||
| ); | ||
|
|
||
| let result; |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify implicit-any declaration still exists at the reported location
rg -n '^\s*let\s+result\s*;' src/lib/neurolink.tsRepository: juspay/neurolink
Length of output: 82
🏁 Script executed:
# Get context around line 1445 to understand result variable usage
sed -n '1430,1460p' src/lib/neurolink.ts | cat -nRepository: juspay/neurolink
Length of output: 1229
🏁 Script executed:
# Find the generate method definition to see its return type
rg -n "^\s*(async\s+)?generate\s*\(" src/lib/neurolink.ts | head -5Repository: juspay/neurolink
Length of output: 82
🏁 Script executed:
# Get the generate method signature and return type
sed -n '2456,2480p' src/lib/neurolink.ts | cat -nRepository: juspay/neurolink
Length of output: 1135
Add explicit type annotation to result variable.
let result; violates strict TypeScript. Since this.generate() returns Promise<GenerateResult>, add the type: let result: GenerateResult;
🧰 Tools
🪛 Biome (2.4.6)
[error] 1445-1445: This variable implicitly has the any type.
(lint/suspicious/noImplicitAnyLet)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/lib/neurolink.ts` at line 1445, The local variable "result" is declared
without a type which breaks strict TypeScript; update the declaration to include
the explicit return type from this.generate() by changing the untyped
declaration to use GenerateResult (i.e., declare result as type GenerateResult).
Locate the variable named result in the method where this.generate() is called
(in src/lib/neurolink.ts) and change its declaration to include the
GenerateResult type so the compiler knows the expected shape.
Author
|
@ coderabbitai review |
Collaborator
|
Needs changes This PR introduces a scheduled task execution system with critical concurrency and resource management flaws that present concrete production risks. Two high-severity issues—non-atomic Redis operations and unbounded key growth—require fixes before deployment.
Also worth fixing:
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
2 times, most recently
from
March 17, 2026 08:27
ca043c8 to
1ad9eff
Compare
Collaborator
|
Changes requested This PR introduces critical concurrency and reliability issues. Non-atomic Redis operations and unhandled rejections risk data loss and crashes under production load.
|
Collaborator
|
Needs changes PR #872 introduces a scheduled task execution system with critical security and deployment issues that must be resolved before merge.
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
from
March 17, 2026 13:34
1ad9eff to
b7b819f
Compare
Collaborator
|
Changes requested PR introduces critical race conditions and runtime crash risks in the scheduled task system. Four critical issues (data corruption under concurrent load and guaranteed crashes on missing optional deps) must be resolved before merge. Critical
High
Medium
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
6 times, most recently
from
March 23, 2026 06:08
e02f842 to
b835642
Compare
Collaborator
|
Changes requested Critical security and runtime failures must be addressed before merge. Dashboard credentials are exposed in logs, optional dependencies cause production crashes, and Redis fallbacks create split-brain scenarios in multi-instance deployments.
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
from
March 23, 2026 11:10
b835642 to
0012e55
Compare
Collaborator
|
needs_changes PR introduces runtime crashes via unconditional optional dependency imports and silent task dropping when delays exceed 24.85 days. High
Medium
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
2 times, most recently
from
March 24, 2026 06:57
9092954 to
e8b0d8f
Compare
Collaborator
|
needs_changes This PR introduces critical reliability risks: silent task drops cause data loss, non-atomic Redis operations create race conditions in multi-worker deployments, and unconditional imports of optional dependencies crash the process when those deps are missing. Critical findings
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
from
March 24, 2026 08:08
e8b0d8f to
db5fdc7
Compare
Collaborator
|
Changes requested This PR introduces critical concurrency defects in the Redis task store. Two non-atomic read-modify-write patterns will corrupt task state and lose run history entries in multi-worker deployments.
|
Collaborator
|
Changes requested PR #872 introduces a scheduled task execution system with two critical defects in the Redis task store: non-atomic operations create race conditions in multi-worker deployments, and missing TTL guarantees unbounded memory growth. Critical findings
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
3 times, most recently
from
March 26, 2026 11:44
f8c5daa to
5a7bf98
Compare
Collaborator
|
needs_changes This PR introduces a scheduled task execution system with critical runtime and security issues that must be resolved before merge. A static import of an optional dependency will crash deployments, while missing authorization controls and credential logging create significant security exposure.
|
Collaborator
|
Changes requested — 6 critical/high severity blocking issues identified. This PR introduces integration crashes, data loss on restart, credential exposure, and cross-instance contamination that must be resolved before merge.
|
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
from
March 26, 2026 13:33
5a7bf98 to
0e0b139
Compare
kanakagrawal-crypto
force-pushed
the
heartbeat-cron-feature
branch
from
March 26, 2026 13:36
0e0b139 to
db5e7a8
Compare
Collaborator
|
needs_changes This PR introduces a scheduled task execution system with critical production risks: Redis KEYS usage causes denial-of-service, and validation gaps enable infinite loops and silent data loss. Blocking
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit
New Features
Chores