fix(cron+providers): Run Now stuck fix, type safety, and claude-cli image format (#577, #568) (#580)

* fix(cron): Run Now executes correctly and reset stale running status (#568)

- PG store RunJob now sets next_run_at = NULL before executeOneJob so
  loadClaimedJob (which requires next_run_at IS NULL) succeeds instead
  of silently skipping the forced run
- recomputeStaleJobs on startup resets last_status 'running' → 'interrupted'
  for jobs that were mid-execution when the server crashed
- UI: add typed CronJobPatch interface matching backend store.CronJobPatch
  (deliverChannel/deliverTo/stateless) — replaces Record<string, unknown>
  on updateJob and onUpdate props across cron detail components
- UI: fix enabled/disabled label in overview tab to reflect local state
  instead of stale job prop

* fix(cron): add error handling for ExecContext calls per code review

- RunJob: return error if claiming job (next_run_at = NULL update) fails
  instead of silently proceeding to executeOneJob which would then skip
- recomputeStaleJobs: log warning on failure, log count of reset jobs
- Add log/slog import to cron_exec.go

* fix(providers): claude-cli ChatSync image format mismatch (#577)

When images are present, Chat() now uses stream-json output format to
match the stream-json input format required by Claude CLI >= v2.1.87.

Also adds concurrent execution guard to RunJob() preventing
double-execution when Run Now is clicked rapidly.

---------

Co-authored-by: viettranx <viettranx@gmail.com>

internal/providers/claude_cli_chat.go

@@ -38,7 +38,14 @@ func (p *ClaudeCLIProvider) Chat(ctx context.Context, req ChatRequest) (*ChatRes
 	disableTools := extractBoolOpt(req.Options, OptDisableTools)
 	bc := bridgeContextFromOpts(req.Options)
 	mcpPath := p.resolveMCPConfigPath(ctx, sessionKey, bc)
-	args := p.buildArgs(model, workDir, mcpPath, cliSessionID, "json", len(images) > 0, disableTools)
+	// Claude CLI >= v2.1.87 requires matching input/output formats.
+	// When images are present, buildArgs adds --input-format stream-json,
+	// so output format must also be stream-json.
+	outputFmt := "json"
+	if len(images) > 0 {
+		outputFmt = "stream-json"
+	}
+	args := p.buildArgs(model, workDir, mcpPath, cliSessionID, outputFmt, len(images) > 0, disableTools)
 
 	var stdin *bytes.Reader
 	if len(images) > 0 {

internal/store/pg/cron_exec.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/google/uuid"
@@ -25,9 +26,20 @@ func (s *PGCronStore) RunJob(ctx context.Context, jobID string, force bool) (boo
 		return false, "", fmt.Errorf("no job handler configured")
 	}
 
-	// Mark job as running before execution
-	if id, parseErr := uuid.Parse(jobID); parseErr == nil {
-		s.db.ExecContext(ctx, "UPDATE cron_jobs SET last_status = 'running', updated_at = $1 WHERE id = $2", time.Now(), id)
+	// Claim the job for forced execution by clearing next_run_at.
+	// executeOneJob calls loadClaimedJob which requires next_run_at IS NULL — same
+	// invariant that claimDueJob establishes for scheduled runs.
+	id, parseErr := uuid.Parse(jobID)
+	if parseErr != nil {
+		return false, "", fmt.Errorf("invalid job id %q: %w", jobID, parseErr)
+	}
+	res, err := s.db.ExecContext(ctx, "UPDATE cron_jobs SET last_status = 'running', next_run_at = NULL, updated_at = $1 WHERE id = $2 AND last_status != 'running'", time.Now(), id)
+	if err != nil {
+		slog.Warn("cron: failed to claim job for forced run", "jobId", jobID, "error", err)
+		return false, "", err
+	}
+	if n, _ := res.RowsAffected(); n == 0 {
+		return false, "", fmt.Errorf("job %s is already running", jobID)
 	}
 	s.mu.Lock()
 	s.cacheLoaded = false

internal/store/pg/cron_scheduler.go

@@ -67,7 +67,17 @@ func (s *PGCronStore) InvalidateCache() {
 // recomputeStaleJobs fixes enabled jobs that have next_run_at = NULL.
 // This happens when the gateway was stopped/crashed while a job was executing,
 // or when the previously computed next_run_at was consumed but never recomputed.
+// Also resets any jobs stuck in 'running' state from a previous crash.
 func (s *PGCronStore) recomputeStaleJobs() {
+	// Reset stale 'running' status — jobs that were mid-execution when the server
+	// crashed will never self-recover, so mark them as interrupted on startup.
+	if res, err := s.db.ExecContext(s.baseCtx,
+		`UPDATE cron_jobs SET last_status = 'interrupted' WHERE last_status = 'running'`); err != nil {
+		slog.Warn("cron: failed to reset stale running jobs on startup", "error", err)
+	} else if n, _ := res.RowsAffected(); n > 0 {
+		slog.Info("cron: reset stale running jobs to interrupted", "count", n)
+	}
+
 	rows, err := s.db.QueryContext(s.baseCtx,
 		`SELECT id, schedule_kind, cron_expression, run_at, timezone, interval_ms
 		 FROM cron_jobs WHERE enabled = true AND next_run_at IS NULL`)

ui/web/src/pages/cron/cron-detail/cron-advanced-dialog.tsx

@@ -9,13 +9,13 @@ import { Dialog, DialogContent, DialogHeader, DialogTitle } from "@/components/u
 import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
 import { ConfigGroupHeader } from "@/components/shared/config-group-header";
 import { IANA_TIMEZONES } from "@/lib/constants";
-import type { CronJob } from "../hooks/use-cron";
+import type { CronJob, CronJobPatch } from "../hooks/use-cron";
 
 interface CronAdvancedDialogProps {
   open: boolean;
   onOpenChange: (open: boolean) => void;
   job: CronJob;
-  onUpdate?: (id: string, params: Record<string, unknown>) => Promise<void>;
+  onUpdate?: (id: string, params: CronJobPatch) => Promise<void>;
 }
 
 function deriveState(job: CronJob) {

ui/web/src/pages/cron/cron-detail/cron-detail-page.tsx

@@ -2,7 +2,7 @@ import { useState, useEffect, useCallback } from "react";
 import { useTranslation } from "react-i18next";
 import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs";
 import { ConfirmDialog } from "@/components/shared/confirm-dialog";
-import type { CronJob, CronRunLogEntry } from "../hooks/use-cron";
+import type { CronJob, CronJobPatch, CronRunLogEntry } from "../hooks/use-cron";
 import { CronHeader } from "./cron-header";
 import { CronOverviewTab } from "./cron-overview-tab";
 import { CronRunHistoryTab } from "./cron-run-history-tab";
@@ -14,7 +14,7 @@ interface CronDetailPageProps {
   onRun: (id: string) => Promise<void>;
   onToggle: (id: string, enabled: boolean) => Promise<void>;
   onDelete: (id: string) => Promise<void>;
-  onUpdate?: (id: string, params: Record<string, unknown>) => Promise<void>;
+  onUpdate?: (id: string, params: CronJobPatch) => Promise<void>;
   getRunLog: (id: string, limit?: number, offset?: number) => Promise<{ entries: CronRunLogEntry[]; total: number }>;
   onRefresh: () => void;
 }

ui/web/src/pages/cron/cron-detail/cron-overview-tab.tsx

@@ -10,13 +10,13 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@
 import { StickySaveBar } from "@/components/shared/sticky-save-bar";
 import { MarkdownRenderer } from "@/components/shared/markdown-renderer";
 import { formatDate } from "@/lib/format";
-import type { CronJob } from "../hooks/use-cron";
+import type { CronJob, CronJobPatch } from "../hooks/use-cron";
 import { CronStatusBadge } from "../cron-utils";
 import { useAgents } from "@/pages/agents/hooks/use-agents";
 
 interface CronOverviewTabProps {
   job: CronJob;
-  onUpdate?: (id: string, params: Record<string, unknown>) => Promise<void>;
+  onUpdate?: (id: string, params: CronJobPatch) => Promise<void>;
 }
 
 type ScheduleKind = "every" | "cron" | "at";
@@ -193,7 +193,7 @@ export function CronOverviewTab({ job, onUpdate }: CronOverviewTabProps) {
           <div className="space-y-2">
             <Label>{t("columns.enabled")}</Label>
             <div className="flex items-center justify-between gap-4 rounded-md border px-3 py-2.5">
-              <span className="text-sm">{job.enabled ? t("detail.enabled") : t("detail.disabled")}</span>
+              <span className="text-sm">{enabled ? t("detail.enabled") : t("detail.disabled")}</span>
               <Switch
                 checked={enabled}
                 onCheckedChange={setEnabled}

ui/web/src/pages/cron/hooks/use-cron.ts

@@ -21,6 +21,20 @@ export interface CronPayload {
   command?: string;
 }
 
+export interface CronJobPatch {
+  name?: string;
+  agentId?: string;
+  enabled?: boolean;
+  schedule?: CronSchedule;
+  message?: string;
+  deliver?: boolean;
+  deliverChannel?: string;
+  deliverTo?: string;
+  deleteAfterRun?: boolean;
+  wakeHeartbeat?: boolean;
+  stateless?: boolean;
+}
+
 export interface CronJob {
   id: string;
   name: string;
@@ -153,7 +167,7 @@ export function useCron() {
   );
 
   const updateJob = useCallback(
-    async (jobId: string, params: Record<string, unknown>) => {
+    async (jobId: string, params: CronJobPatch) => {
       try {
         await ws.call(Methods.CRON_UPDATE, { jobId, patch: params });
         await invalidate();