feat: per-connector schedules (targeted cron pause/resume/delete)

src/agent/okf-middleware.ts

@@ -28,13 +28,38 @@ export function createOpenWikiIndexMiddleware(
     beforeAgent: async () => {
       await migrateWikiToOkf(backend, outputMode);
     },
-    wrapToolCall: async (request, handler) =>
-      addFrontmatterWarning(
-        await handler(request),
+    wrapToolCall: async (request, handler) => {
+      // Let the tool execute first. If the tool itself throws, the error
+      // propagates through LangChain's composition layer as a
+      // MiddlewareError whose root cause is NOT a ToolInvocationError.
+      // ToolNode.#handleError then re-throws it as fatal because
+      // handleToolErrors !== true. To avoid that crash path, we catch
+      // tool execution errors here and convert them to ToolMessages so
+      // the LLM can see the failure and retry.
+      let result: unknown;
+      try {
+        result = await handler(request);
+      } catch (error) {
+        // Re-throw GraphInterrupt and abort signals untouched — those are
+        // control flow, not tool failures.
+        if (isGraphInterruptLike(error) || isAborted(error)) {
+          throw error;
+        }
+        // Convert the tool error into a ToolMessage. This mirrors
+        // LangChain's defaultHandleToolErrors but runs before the
+        // composition layer wraps it in MiddlewareError.
+        return toolErrorToMessage(error, request.toolCall);
+      }
+
+      // Only post-process successful results — frontmatter validation
+      // must not run on error results (which are already ToolMessages).
+      return addFrontmatterWarning(
+        result,
         backend,
         outputMode,
         request.toolCall.name,
-      ),
+      );
+    },
     afterAgent: async () => {
       await validateWikiMermaid(backend, outputMode);
       await synchronizeWikiIndexes(backend, outputMode);
@@ -44,6 +69,10 @@ export function createOpenWikiIndexMiddleware(
 
 /**
  * Appends an actionable warning when a wiki write leaves invalid front matter.
+ *
+ * Errors from `validatePersistedFile` are caught and swallowed so a
+ * validation failure never crashes the agent run — the original tool
+ * result is returned unchanged.
  */
 export async function addFrontmatterWarning<Result>(
   result: Result,
@@ -65,14 +94,20 @@ export async function addFrontmatterWarning<Result>(
     );
   if (!mutation) return result;
 
-  const validation = await validatePersistedFile(backend, mutation.path);
-  if (validation.valid) return result;
+  try {
+    const validation = await validatePersistedFile(backend, mutation.path);
+    if (validation.valid) return result;
+
+    const warning = formatWarning(mutation.path, validation.issues);
+    mutation.message.content =
+      typeof mutation.message.content === "string"
+        ? `${mutation.message.content}\n\n${warning}`
+        : [...mutation.message.content, { text: warning, type: "text" }];
+  } catch {
+    // Swallow validation errors — a failed frontmatter check must not
+    // prevent the tool result from reaching the LLM.
+  }
 
-  const warning = formatWarning(mutation.path, validation.issues);
-  mutation.message.content =
-    typeof mutation.message.content === "string"
-      ? `${mutation.message.content}\n\n${warning}`
-      : [...mutation.message.content, { text: warning, type: "text" }];
   return result;
 }
 
@@ -127,3 +162,41 @@ function formatWarning(path: string, issues: FrontmatterIssue[]): string {
 function isRecord(value: unknown): value is Record<string, unknown> {
   return value !== null && typeof value === "object" && !Array.isArray(value);
 }
+
+/**
+ * Returns true when the error is a LangGraph interrupt (graph-level control
+ * flow that must propagate untouched).
+ */
+function isGraphInterruptLike(error: unknown): boolean {
+  if (!isRecord(error)) return false;
+  // LangGraph Interrupt errors carry a `__interrupt` symbol or type marker.
+  if (error.__interrupt === true) return true;
+  if (typeof error.type === "string" && error.type === "interrupt") return true;
+  return false;
+}
+
+/**
+ * Returns true when the error signals an aborted run.
+ */
+function isAborted(error: unknown): boolean {
+  if (!isRecord(error)) return false;
+  if (error.name === "AbortError" || error.name === "CancelledError") return true;
+  return false;
+}
+
+/**
+ * Converts a thrown tool error into a ToolMessage so the LLM sees the
+ * failure message and can retry, instead of the process crashing.
+ */
+function toolErrorToMessage(
+  error: unknown,
+  toolCall: { id: string; name: string },
+): ToolMessage {
+  const message =
+    error instanceof Error ? error.message : String(error);
+  return new ToolMessage({
+    content: `${message}\n Please fix your mistakes.`,
+    tool_call_id: toolCall.id,
+    name: toolCall.name,
+  });
+}

src/cli.tsx

@@ -3748,6 +3748,22 @@ async function runCronCommand(
         throw new Error(`Target is required for cron ${command.action}.`);
       }
 
+      // Validate that a source-instance target actually exists in the config.
+      if (
+        typeof command.target === "object" &&
+        command.target.kind === "source-instance"
+      ) {
+        const targetId = command.target.id;
+        const instanceExists = config.sourceInstances.some(
+          (s) => s.id === targetId,
+        );
+        if (!instanceExists) {
+          throw new Error(
+            `No source instance found with ID "${targetId}".`,
+          );
+        }
+      }
+
       const result =
         command.action === "pause"
           ? await pauseConnectorSchedules(config, command.target)
@@ -3795,17 +3811,34 @@ async function printCronSchedules(
   }
 }
 
+function resolveSourceInstanceLabel(
+  sourceInstanceId: string,
+  sourceInstances: { id: string; name?: string; connectorId: string }[],
+): string {
+  const source = sourceInstances.find((s) => s.id === sourceInstanceId);
+  if (source?.name) return source.name;
+  if (source?.connectorId) return source.connectorId;
+  return sourceInstanceId;
+}
+
 function formatScheduleMutationResult(
   action: "delete" | "pause" | "resume",
   result: ScheduleMutationResult,
 ): string {
   const actionLabel =
     action === "delete" ? "Deleted" : action === "pause" ? "Paused" : "Resumed";
+  const sourceInstances = result.config.sourceInstances;
   const changed =
-    result.connectorIds.length > 0 ? result.connectorIds.join(", ") : "none";
+    result.connectorIds.length > 0
+      ? result.connectorIds
+          .map((id) => resolveSourceInstanceLabel(id, sourceInstances))
+          .join(", ")
+      : "none";
   const skipped =
     result.skippedConnectorIds.length > 0
-      ? result.skippedConnectorIds.join(", ")
+      ? result.skippedConnectorIds
+          .map((id) => resolveSourceInstanceLabel(id, sourceInstances))
+          .join(", ")
       : "none";
   const rows = [
     [`${actionLabel}`, changed],

src/commands.ts

@@ -10,7 +10,7 @@ export type HelpRow = {
 };
 
 export type OpenWikiRunMode = "personal" | "code";
-type CronTarget = Extract<IngestionTarget, string>;
+type CronTarget = IngestionTarget;
 
 export type HelpContent = {
   title: string;
@@ -295,12 +295,20 @@ export function parseCommand(argv: string[]): CliCommand {
     }
 
     if (argv[1] === "pause" || argv[1] === "resume" || argv[1] === "delete") {
+      if (argv.length > 3) {
+        return {
+          kind: "error",
+          exitCode: 1,
+          message: `Usage: openwiki cron ${argv[1]} <source|source-instance|all>`,
+        };
+      }
+
       const target = parseIngestionTarget(argv[2] ?? "");
-      if (target !== "all" || argv.length > 3) {
+      if (!target) {
         return {
           kind: "error",
           exitCode: 1,
-          message: `Usage: openwiki cron ${argv[1]} all`,
+          message: `Usage: openwiki cron ${argv[1]} <source|source-instance|all>`,
         };
       }
 
@@ -317,7 +325,7 @@ export function parseCommand(argv: string[]): CliCommand {
         kind: "error",
         exitCode: 1,
         message:
-          "Usage: openwiki cron list | pause all | resume all | delete all",
+          "Usage: openwiki cron <list|pause|resume|delete> [<source|source-instance|all>]",
       };
     }
   }
@@ -649,9 +657,9 @@ export const helpContent: HelpContent = {
     "openwiki auth tools <provider>",
     "openwiki ingest <source|source-instance|all> [--scheduled] [--print] [--modelId <id>]",
     "openwiki cron list",
-    "openwiki cron pause all",
-    "openwiki cron resume all",
-    "openwiki cron delete all",
+    "openwiki cron pause <source|source-instance|all>",
+    "openwiki cron resume <source|source-instance|all>",
+    "openwiki cron delete <source|source-instance|all>",
     "openwiki ngrok start [url] [--port <port>]",
   ],
   commands: [
@@ -694,19 +702,19 @@ export const helpContent: HelpContent = {
       description: "List saved connector schedules and local launchd status.",
     },
     {
-      label: "openwiki cron pause all",
+      label: "openwiki cron pause <source|source-instance|all>",
       description:
-        "Pause saved connector schedules and reconcile the Mac wake window.",
+        "Pause saved connector schedules for a connector, source instance, or all sources.",
     },
     {
-      label: "openwiki cron resume all",
+      label: "openwiki cron resume <source|source-instance|all>",
       description:
-        "Resume paused connector schedules and reconcile the Mac wake window.",
+        "Resume paused connector schedules for a connector, source instance, or all sources.",
     },
     {
-      label: "openwiki cron delete all",
+      label: "openwiki cron delete <source|source-instance|all>",
       description:
-        "Delete saved connector schedules and remove stale local schedule files.",
+        "Delete saved connector schedules for a connector, source instance, or all sources.",
     },
     {
       label: "openwiki ngrok start [url]",
@@ -778,8 +786,11 @@ export const helpContent: HelpContent = {
     "openwiki ingest web-search-2",
     "openwiki cron list",
     "openwiki cron pause all",
+    "openwiki cron pause web-search",
     "openwiki cron resume all",
+    "openwiki cron resume web-search",
     "openwiki cron delete all",
+    "openwiki cron delete web-search",
     "openwiki auth slack",
     "openwiki auth gmail",
     "openwiki auth notion",

src/credentials.tsx

@@ -2574,6 +2574,7 @@ export function InitSetup({
         connectorId: "git-repo",
         cronExpression,
         cwd: process.cwd(),
+        sourceInstanceId: "git-repo",
       });
       const nextConfig: OpenWikiOnboardingConfig = {
         ...onboardingConfig,

src/ingestion.ts

@@ -213,11 +213,15 @@ function resolveIngestionSourceInstances(
       return false;
     }
 
-    if (
-      scheduledOnly &&
-      (!config.ingestionSchedule || config.ingestionSchedule.pausedAt)
-    ) {
-      return false;
+    if (scheduledOnly) {
+      // If no per-source schedule exists, the source is not scheduled — exclude it
+      if (!sourceConfig.schedule) {
+        return false;
+      }
+      // If the per-source schedule is paused, exclude it
+      if (sourceConfig.schedule.pausedAt) {
+        return false;
+      }
     }
 
     if (target === "all") {

src/onboarding.ts

@@ -344,17 +344,22 @@ function normalizeOnboardingConfig(value: unknown): OpenWikiOnboardingConfig {
     }
   }
 
-  if (!config.ingestionSchedule) {
+  // Migrate global ingestionSchedule to source instances that lack their own schedule.
+  // Per-source schedules are the source of truth; the global is kept for backward compatibility.
+  if (config.ingestionSchedule) {
+    config.sourceInstances = config.sourceInstances.map((sourceConfig) => {
+      if (sourceConfig.schedule) {
+        return sourceConfig;
+      }
+      return { ...sourceConfig, schedule: config.ingestionSchedule };
+    });
+  } else {
+    // Backfill global ingestionSchedule from first source instance that has one.
     config.ingestionSchedule = config.sourceInstances.find(
       (sourceConfig) => sourceConfig.schedule,
     )?.schedule;
   }
 
-  config.sourceInstances = config.sourceInstances.map((sourceConfig) => {
-    const nextSourceConfig = { ...sourceConfig };
-    delete nextSourceConfig.schedule;
-    return nextSourceConfig;
-  });
   config.sources = deriveLegacySources(config.sourceInstances);
 
   return config;
@@ -411,6 +416,7 @@ function deriveLegacySources(
         connectedAt: sourceInstance.connectedAt,
         connectorConfig: sourceInstance.connectorConfig,
         ingestionGoal: sourceInstance.ingestionGoal,
+        schedule: sourceInstance.schedule,
       };
     }
   }