diff --git a/internal/providers/claude_cli_chat.go b/internal/providers/claude_cli_chat.go index 7d7bacc3..cb8ec571 100644 --- a/internal/providers/claude_cli_chat.go +++ b/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 { diff --git a/internal/store/pg/cron_exec.go b/internal/store/pg/cron_exec.go index 514090c1..ff139b01 100644 --- a/internal/store/pg/cron_exec.go +++ b/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 diff --git a/internal/store/pg/cron_scheduler.go b/internal/store/pg/cron_scheduler.go index 9101a601..18b64efa 100644 --- a/internal/store/pg/cron_scheduler.go +++ b/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`) diff --git a/ui/web/src/pages/cron/cron-detail/cron-advanced-dialog.tsx b/ui/web/src/pages/cron/cron-detail/cron-advanced-dialog.tsx index f99bd842..de58ac73 100644 --- a/ui/web/src/pages/cron/cron-detail/cron-advanced-dialog.tsx +++ b/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) => Promise; + onUpdate?: (id: string, params: CronJobPatch) => Promise; } function deriveState(job: CronJob) { diff --git a/ui/web/src/pages/cron/cron-detail/cron-detail-page.tsx b/ui/web/src/pages/cron/cron-detail/cron-detail-page.tsx index 1db0450f..f9f84b67 100644 --- a/ui/web/src/pages/cron/cron-detail/cron-detail-page.tsx +++ b/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; onToggle: (id: string, enabled: boolean) => Promise; onDelete: (id: string) => Promise; - onUpdate?: (id: string, params: Record) => Promise; + onUpdate?: (id: string, params: CronJobPatch) => Promise; getRunLog: (id: string, limit?: number, offset?: number) => Promise<{ entries: CronRunLogEntry[]; total: number }>; onRefresh: () => void; } diff --git a/ui/web/src/pages/cron/cron-detail/cron-overview-tab.tsx b/ui/web/src/pages/cron/cron-detail/cron-overview-tab.tsx index df461ddd..01774bdd 100644 --- a/ui/web/src/pages/cron/cron-detail/cron-overview-tab.tsx +++ b/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) => Promise; + onUpdate?: (id: string, params: CronJobPatch) => Promise; } type ScheduleKind = "every" | "cron" | "at"; @@ -193,7 +193,7 @@ export function CronOverviewTab({ job, onUpdate }: CronOverviewTabProps) {
- {job.enabled ? t("detail.enabled") : t("detail.disabled")} + {enabled ? t("detail.enabled") : t("detail.disabled")} ) => { + async (jobId: string, params: CronJobPatch) => { try { await ws.call(Methods.CRON_UPDATE, { jobId, patch: params }); await invalidate();